41 lines
1.7 KiB
Python
41 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MIGRATION = ROOT / "src/dotnet/QuantEngine.Infrastructure/Migrations/V6__Add_Market_Time_Series.sql"
|
|
DBML = ROOT / "docs/db/quantengine.dbml"
|
|
REPORT = ROOT / "Temp/market_time_series_schema_v1.json"
|
|
TABLES = ("quantengine.price_history_daily", "quantengine.macro_history_daily")
|
|
|
|
|
|
def main() -> int:
|
|
migration = MIGRATION.read_text(encoding="utf-8") if MIGRATION.exists() else ""
|
|
dbml = DBML.read_text(encoding="utf-8") if DBML.exists() else ""
|
|
checks = {
|
|
"migration_exists": MIGRATION.exists(),
|
|
"dbml_exists": DBML.exists(),
|
|
"migration_tables": all(re.search(rf"CREATE TABLE IF NOT EXISTS {re.escape(t)}\b", migration) for t in TABLES),
|
|
"dbml_tables": all(re.search(rf"Table {re.escape(t)}\s*\{{", dbml) for t in TABLES),
|
|
"price_primary_key": "PRIMARY KEY (ticker, trade_date)" in migration,
|
|
"macro_primary_key": "PRIMARY KEY (symbol, trade_date)" in migration,
|
|
"dbup_embedded_glob": "Migrations/**/*.sql" in (ROOT / "src/dotnet/QuantEngine.Infrastructure/QuantEngine.Infrastructure.csproj").read_text(encoding="utf-8"),
|
|
}
|
|
payload = {
|
|
"formula_id": "MARKET_TIME_SERIES_SCHEMA_V1",
|
|
"gate": "PASS" if all(checks.values()) else "FAIL",
|
|
"expected_tables": list(TABLES),
|
|
"checks": checks,
|
|
"runtime_database_query": "DATA_GATED",
|
|
}
|
|
REPORT.parent.mkdir(parents=True, exist_ok=True)
|
|
REPORT.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
return 0 if payload["gate"] == "PASS" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|