#!/usr/bin/env python3 from __future__ import annotations import json import os import sys from datetime import date, timedelta from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] REPORT = ROOT / "Temp" / "price_history_integrity_v1.json" # Import db connection resolution (inlined to avoid fragility) try: import psycopg except ImportError: psycopg = None try: import yaml except ImportError: yaml = None # Import trading calendar sys.path.insert(0, str(ROOT / "src")) try: from quant_engine.lib_trading_calendar import is_trading_day except ImportError: is_trading_day = None def parse_dotnet_connection_string(s: str) -> dict[str, str | None]: """Parse .NET connection string format to psycopg-compatible dict.""" result: dict[str, str | None] = {} parts = s.split(";") search_path_value = None for part in parts: part = part.strip() if not part or "=" not in part: continue key, value = part.split("=", 1) key_lower = key.strip().lower() value = value.strip() if key_lower == "host": result["host"] = value elif key_lower == "port": result["port"] = value elif key_lower == "database": result["dbname"] = value elif key_lower == "username": result["user"] = value elif key_lower == "password": result["password"] = value elif key_lower == "search path": search_path_value = value if search_path_value: result["options"] = f"-c search_path={search_path_value}" return result def _build_psycopg_dsn(parsed: dict[str, str | None]) -> str: """Build psycopg DSN from parsed dict.""" parts = [] for key in ["host", "port", "dbname", "user", "password"]: val = parsed.get(key) if val: parts.append(f"{key}={val}") return " ".join(parts) def resolve_db_connection() -> str | None: """Resolve PostgreSQL connection string. Resolution order: 1. env QE_WBS_PG_DSN (psycopg DSN format) 2. env ConnectionStrings__DefaultConnection (.NET format, convert to psycopg) 3. appsettings.Development.json ConnectionStrings.DefaultConnection (.NET format, convert) """ # Try env QE_WBS_PG_DSN dsn = os.environ.get("QE_WBS_PG_DSN") if dsn: return dsn # Try env ConnectionStrings__DefaultConnection (.NET format) dotnet_str = os.environ.get("ConnectionStrings__DefaultConnection") if dotnet_str: parsed = parse_dotnet_connection_string(dotnet_str) return _build_psycopg_dsn(parsed) # Try appsettings.Development.json appsettings_path = ROOT / "src/dotnet/QuantEngine.Web/appsettings.Development.json" if appsettings_path.exists(): try: appsettings = json.loads(appsettings_path.read_text(encoding="utf-8")) conn_str = appsettings.get("ConnectionStrings", {}).get("DefaultConnection", "") if conn_str: parsed = parse_dotnet_connection_string(conn_str) return _build_psycopg_dsn(parsed) except Exception: pass return None def query_price_data(dsn: str) -> tuple[bool, dict[str, list[tuple[str, date]]] | str]: """Query price_history_daily table. Returns (success, data_or_error_msg). On success: data = {ticker: [(ticker, trade_date), ...], ...} On error: error message string. """ if psycopg is None: return False, "psycopg not installed" try: conn = psycopg.connect(dsn) try: cursor = conn.cursor() # Query tickers and dates, ordered for easier grouping cursor.execute( "SELECT ticker, trade_date FROM quantengine.price_history_daily ORDER BY ticker, trade_date" ) rows = cursor.fetchall() cursor.close() # Group by ticker data: dict[str, list[tuple[str, date]]] = {} for ticker, trade_date in rows: if ticker not in data: data[ticker] = [] data[ticker].append((ticker, trade_date)) return True, data except Exception as e: return False, str(e) finally: conn.close() except Exception as e: return False, str(e) def query_price_sanity(dsn: str) -> tuple[bool, int | str]: """Query for invalid price rows. Returns (success, count_or_error_msg). """ if psycopg is None: return False, "psycopg not installed" try: conn = psycopg.connect(dsn) try: cursor = conn.cursor() # Count rows with invalid OHLCV cursor.execute( """SELECT COUNT(*) FROM quantengine.price_history_daily WHERE open <= 0 OR high <= 0 OR low <= 0 OR close <= 0 OR volume < 0""" ) row = cursor.fetchone() cursor.close() count = row[0] if row else 0 return True, count except Exception as e: return False, str(e) finally: conn.close() except Exception as e: return False, str(e) def compute_gaps(data: dict[str, list[tuple[str, date]]]) -> tuple[int, dict[str, Any]]: """Compute gap analysis per ticker. Returns (total_gap_count, per_ticker_details). """ if is_trading_day is None: # Can't compute gaps without trading calendar return 0, {} total_gaps = 0 per_ticker = [] for ticker, entries in sorted(data.items()): if not entries: continue trade_dates = sorted(set(t[1] for t in entries)) min_date = trade_dates[0] max_date = trade_dates[-1] # Iterate through all dates in range and count missing trading days missing_trading_days = 0 current_date = min_date while current_date <= max_date: if is_trading_day(current_date) and current_date not in trade_dates: missing_trading_days += 1 current_date += timedelta(days=1) total_gaps += missing_trading_days per_ticker.append({ "ticker": ticker, "min_date": min_date.isoformat(), "max_date": max_date.isoformat(), "row_count": len(entries), "missing_trading_days": missing_trading_days }) return total_gaps, per_ticker def main() -> int: """Main entry point.""" dsn = resolve_db_connection() # Attempt DB queries if not dsn: payload = { "formula_id": "PRICE_HISTORY_INTEGRITY_V1", "gate": "FAIL", "error": "No PostgreSQL connection available (QE_WBS_PG_DSN or ConnectionStrings__DefaultConnection env var not set, and appsettings.Development.json not found or not accessible)", "gap_count": None, "invalid_price_rows": None, "scope_note": "gap-freeness checked within each ticker's currently-collected min~max trade_date range; does not assert full historical coverage (see QE-M2-03 for backfill)", "per_ticker": [] } 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 1 # Query price data success, result = query_price_data(dsn) if not success: payload = { "formula_id": "PRICE_HISTORY_INTEGRITY_V1", "gate": "FAIL", "error": f"Failed to query price_history_daily: {result}", "gap_count": None, "invalid_price_rows": None, "scope_note": "gap-freeness checked within each ticker's currently-collected min~max trade_date range; does not assert full historical coverage (see QE-M2-03 for backfill)", "per_ticker": [] } 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 1 price_data = result if is_trading_day is None: payload = { "formula_id": "PRICE_HISTORY_INTEGRITY_V1", "gate": "FAIL", "error": "Failed to import is_trading_day from quant_engine.lib_trading_calendar", "gap_count": None, "invalid_price_rows": None, "scope_note": "gap-freeness checked within each ticker's currently-collected min~max trade_date range; does not assert full historical coverage (see QE-M2-03 for backfill)", "per_ticker": [] } 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 1 # Query price sanity success, result = query_price_sanity(dsn) if not success: payload = { "formula_id": "PRICE_HISTORY_INTEGRITY_V1", "gate": "FAIL", "error": f"Failed to check price sanity: {result}", "gap_count": None, "invalid_price_rows": None, "scope_note": "gap-freeness checked within each ticker's currently-collected min~max trade_date range; does not assert full historical coverage (see QE-M2-03 for backfill)", "per_ticker": [] } 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 1 invalid_price_rows = result # Compute gaps gap_count, per_ticker_details = compute_gaps(price_data) # Determine gate result gate = "PASS" if gap_count == 0 and invalid_price_rows == 0 else "FAIL" # Build payload payload = { "formula_id": "PRICE_HISTORY_INTEGRITY_V1", "gate": gate, "gap_count": gap_count, "invalid_price_rows": invalid_price_rows, "scope_note": "gap-freeness checked within each ticker's currently-collected min~max trade_date range; does not assert full historical coverage (see QE-M2-03 for backfill)", "per_ticker": per_ticker_details } # Write report REPORT.parent.mkdir(parents=True, exist_ok=True) REPORT.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") # Print JSON to stdout print(json.dumps(payload, ensure_ascii=False, indent=2)) # Exit with appropriate code return 0 if gate == "PASS" else 1 if __name__ == "__main__": raise SystemExit(main())