QuantEngineByItz/tools/validate_price_history_integrity_v1.py

#!/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())