OpenFin/backend/services/database.py

"""
SQLite persistence layer for portfolio positions, custom patterns, and config.
"""
import sqlite3
import json
import os
from datetime import datetime
from typing import List, Dict, Any, Optional

DB_PATH = os.path.join(os.path.dirname(__file__), "..", "data", "geooptions.db")


def get_conn() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
    conn = get_conn()
    c = conn.cursor()

    c.execute("""CREATE TABLE IF NOT EXISTS portfolio (
        id TEXT PRIMARY KEY,
        title TEXT NOT NULL,
        underlying TEXT NOT NULL,
        strategy TEXT NOT NULL,
        asset_class TEXT,
        entry_date TEXT NOT NULL,
        expiry_date TEXT,
        expiry_days INTEGER,
        legs TEXT NOT NULL,
        capital_invested REAL NOT NULL,
        entry_underlying_price REAL,
        geo_trigger TEXT,
        rationale TEXT,
        status TEXT DEFAULT 'open',
        close_date TEXT,
        close_value REAL,
        notes TEXT,
        ib_fees_entry REAL DEFAULT 0,
        ib_fees_exit REAL DEFAULT 0,
        created_at TEXT DEFAULT (datetime('now'))
    )""")

    c.execute("""CREATE TABLE IF NOT EXISTS custom_patterns (
        id TEXT PRIMARY KEY,
        name TEXT NOT NULL,
        description TEXT,
        triggers TEXT,
        keywords TEXT,
        historical_instances TEXT,
        suggested_trades TEXT,
        asset_class TEXT,
        expected_move_pct REAL,
        probability REAL,
        horizon_days INTEGER,
        ai_quality_score INTEGER,
        ai_evaluation TEXT,
        source TEXT DEFAULT 'custom',
        is_active INTEGER DEFAULT 1,
        created_at TEXT DEFAULT (datetime('now')),
        updated_at TEXT DEFAULT (datetime('now'))
    )""")
    # Migrations: add columns if not present
    for _sql in [
        "ALTER TABLE custom_patterns ADD COLUMN source TEXT DEFAULT 'custom'",
        "ALTER TABLE custom_patterns ADD COLUMN counter_thesis TEXT",
        "ALTER TABLE custom_patterns ADD COLUMN invalidation_trigger TEXT",
        "ALTER TABLE custom_patterns ADD COLUMN invalidation_probability REAL",
        "ALTER TABLE pattern_score_history ADD COLUMN predicted_probability REAL",
        "ALTER TABLE knowledge_base ADD COLUMN expires_at TEXT",
        "ALTER TABLE knowledge_base ADD COLUMN confidence_decay_days INTEGER DEFAULT 90",
        # Phase 4.1 — Bayesian posteriors
        "ALTER TABLE custom_patterns ADD COLUMN bayesian_alpha REAL DEFAULT 1.0",
        "ALTER TABLE custom_patterns ADD COLUMN bayesian_beta REAL DEFAULT 1.0",
        "ALTER TABLE custom_patterns ADD COLUMN bayesian_win_rate REAL",
        "ALTER TABLE custom_patterns ADD COLUMN bayesian_updated_at TEXT",
        "ALTER TABLE custom_patterns ADD COLUMN bayesian_sample_size INTEGER DEFAULT 0",
        # Convergence Phase 1 — thematic classification
        "ALTER TABLE custom_patterns ADD COLUMN category TEXT",
        "ALTER TABLE custom_patterns ADD COLUMN signal_direction TEXT",
        # Pattern Explorer — taxonomy tree path
        "ALTER TABLE custom_patterns ADD COLUMN taxonomy_path TEXT DEFAULT '[]'",
    ]:
        try:
            c.execute(_sql)
        except Exception:
            pass

    # Phase 4.2 — Régime clusters (K-Means sur gauges macro)
    c.execute("""CREATE TABLE IF NOT EXISTS regime_clusters (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        timestamp TEXT NOT NULL,
        cluster_id INTEGER NOT NULL,
        cluster_label TEXT,
        dominant_regime TEXT,
        gauges_json TEXT DEFAULT '{}',
        anomaly_flag INTEGER DEFAULT 0,
        created_at TEXT DEFAULT (datetime('now'))
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_rc_ts ON regime_clusters(timestamp DESC)")
    except Exception:
        pass

    # Phase 4.3 — Embeddings sémantiques des patterns
    c.execute("""CREATE TABLE IF NOT EXISTS pattern_embeddings (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        pattern_id TEXT NOT NULL UNIQUE,
        embedding_json TEXT NOT NULL,
        model_version TEXT DEFAULT 'text-embedding-3-small',
        created_at TEXT DEFAULT (datetime('now')),
        updated_at TEXT DEFAULT (datetime('now'))
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_pe_pattern ON pattern_embeddings(pattern_id)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS config (
        key TEXT PRIMARY KEY,
        value TEXT NOT NULL,
        updated_at TEXT DEFAULT (datetime('now'))
    )""")

    c.execute("""CREATE TABLE IF NOT EXISTS pattern_score_history (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        run_id TEXT NOT NULL,
        pattern_id TEXT NOT NULL,
        score INTEGER,
        confidence INTEGER,
        summary TEXT,
        scored_at TEXT NOT NULL
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_psh_pattern ON pattern_score_history(pattern_id, scored_at DESC)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS cycle_runs (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        run_id TEXT NOT NULL UNIQUE,
        started_at TEXT NOT NULL,
        completed_at TEXT,
        trigger TEXT DEFAULT 'auto',
        patterns_suggested INTEGER DEFAULT 0,
        patterns_added INTEGER DEFAULT 0,
        patterns_scored INTEGER DEFAULT 0,
        geo_score INTEGER,
        dominant_regime TEXT,
        commentary TEXT,
        status TEXT DEFAULT 'running'
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_cr_started ON cycle_runs(started_at DESC)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS macro_regime_history (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        timestamp TEXT NOT NULL,
        dominant TEXT NOT NULL,
        scores_json TEXT NOT NULL,
        reasons_json TEXT NOT NULL,
        gauges_summary_json TEXT NOT NULL DEFAULT '{}'
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_mrh_ts ON macro_regime_history(timestamp DESC)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS geo_alert_history (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        timestamp TEXT NOT NULL,
        run_id TEXT NOT NULL,
        geo_score INTEGER NOT NULL,
        top_patterns_json TEXT NOT NULL DEFAULT '[]',
        news_count INTEGER DEFAULT 0
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_gah_ts ON geo_alert_history(timestamp DESC)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS trade_entry_prices (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        run_id TEXT NOT NULL,
        pattern_id TEXT NOT NULL,
        pattern_name TEXT,
        underlying TEXT NOT NULL,
        strategy TEXT,
        entry_price REAL,
        entry_date TEXT NOT NULL,
        score_at_entry INTEGER DEFAULT 0,
        latest_score INTEGER,
        expected_move_pct REAL,
        horizon_days INTEGER DEFAULT 30,
        ev_at_entry REAL,
        ev_net REAL,
        trade_score REAL,
        matched_profile TEXT,
        last_seen_at TEXT
    )""")
    for col, definition in [
        ("latest_score", "INTEGER"),
        ("ev_at_entry", "REAL"),
        ("ev_net", "REAL"),
        ("trade_score", "REAL"),
        ("matched_profile", "TEXT"),
        ("last_seen_at", "TEXT"),
        ("pnl_pct", "REAL"),
        ("capital_invested", "REAL"),
        ("strike_guidance", "TEXT"),
        ("expiry_days_at_entry", "INTEGER"),
        ("status", "TEXT DEFAULT 'open'"),
        ("closed_at", "TEXT"),
        ("close_reason", "TEXT"),
        ("close_note", "TEXT"),
        ("pnl_realized", "REAL"),
        ("close_price", "REAL"),
        ("target_pct", "REAL"),
        ("stop_loss_pct", "REAL"),
        ("signal_threshold", "REAL"),
        ("asset_class", "TEXT"),
    ]:
        try:
            c.execute(f"ALTER TABLE trade_entry_prices ADD COLUMN {col} {definition}")
        except Exception:
            pass
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_tep_date ON trade_entry_prices(entry_date DESC)")
        c.execute("CREATE INDEX IF NOT EXISTS idx_tep_status ON trade_entry_prices(status, closed_at DESC)")
    except Exception:
        pass

    # One-time migration: fix existing NULL asset_class rows using ticker lookup
    _backfill_cases = [
        ("energy",      "'CL=F','BZ=F','NG=F','RB=F','HO=F','XLE','XOP','USO','UCO','BOIL','UNG'"),
        ("metals",      "'GC=F','SI=F','HG=F','PA=F','PL=F','GLD','IAU','SLV','GDX','GDXJ','GLDM','PPLT'"),
        ("agriculture", "'ZC=F','ZS=F','ZW=F','CC=F','KC=F','CT=F','OJ=F','CORN','WEAT','SOYB','DBA'"),
        ("indices",     "'^GSPC','^NDX','^DJI','^RUT','SPY','QQQ','IWM','DIA','RSP','VGK','EEM','EWZ','FXI','EWJ','EFA','ACWI','INDA','^NSEI','^HSI','EWG','EWU','EWI'"),
        ("equities",    "'XLF','XLK','XLV','XLI','XLP','XLU','XLY','XLRE','XLB','XLC'"),
        ("forex",       "'DX-Y.NYB','UUP','FXE','FXY','EUO','YCS','FXA','FXB','FXF'"),
    ]
    for _cls, _tickers in _backfill_cases:
        try:
            c.execute(
                f"UPDATE trade_entry_prices SET asset_class=? WHERE (asset_class IS NULL OR asset_class='') AND underlying IN ({_tickers})",
                (_cls,)
            )
            c.execute(
                f"UPDATE skipped_trades SET asset_class=? WHERE (asset_class IS NULL OR asset_class='') AND underlying IN ({_tickers})",
                (_cls,)
            )
        except Exception:
            pass

    c.execute("""CREATE TABLE IF NOT EXISTS risk_profiles (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        min_score INTEGER NOT NULL DEFAULT 0,
        min_gain_pct REAL NOT NULL DEFAULT 0,
        color TEXT DEFAULT '#3b82f6',
        enabled INTEGER DEFAULT 1,
        sort_order INTEGER DEFAULT 0,
        created_at TEXT DEFAULT (datetime('now'))
    )""")

    # Default risk profiles (seed only if table is empty)
    existing_profiles = c.execute("SELECT COUNT(*) FROM risk_profiles").fetchone()[0]
    if existing_profiles == 0:
        default_profiles = [
            ("Conservateur", 50, 100.0, "#22c55e", 1, 0),
            ("Équilibré",    30, 250.0, "#3b82f6", 1, 1),
            ("Agressif",     15, 600.0, "#ef4444", 1, 2),
        ]
        c.executemany(
            "INSERT INTO risk_profiles (name, min_score, min_gain_pct, color, enabled, sort_order) VALUES (?,?,?,?,?,?)",
            default_profiles
        )

    # Default config
    defaults = {
        "openai_api_key": "",
        "newsapi_key": "",
        "eia_api_key": "",
        "fred_api_key": "",
        "sources": json.dumps({
            "reuters_world": {"enabled": True, "url": "https://feeds.reuters.com/reuters/worldNews", "name": "Reuters World"},
            "reuters_business": {"enabled": True, "url": "https://feeds.reuters.com/reuters/businessNews", "name": "Reuters Business"},
            "reuters_energy": {"enabled": True, "url": "https://feeds.reuters.com/reuters/USenergyNews", "name": "Reuters Commodities"},
            "ap_top": {"enabled": True, "url": "https://feeds.apnews.com/rss/apf-topnews", "name": "AP Top News"},
            "aljazeera": {"enabled": True, "url": "https://www.aljazeera.com/xml/rss/all.xml", "name": "Al Jazeera"},
            "ft": {"enabled": False, "url": "https://www.ft.com/rss/home", "name": "Financial Times"},
            "bloomberg": {"enabled": False, "url": "https://feeds.bloomberg.com/markets/news.rss", "name": "Bloomberg Markets"},
            "newsapi": {"enabled": False, "url": "", "name": "NewsAPI (clé requise)", "requires_key": "newsapi_key"},
            "gdelt": {"enabled": False, "url": "https://api.gdeltproject.org/api/v2/doc/doc?query=geopolitics&mode=artlist&format=json", "name": "GDELT Project (gratuit)"},
            "eia": {"enabled": False, "url": "", "name": "EIA Energy (clé requise)", "requires_key": "eia_api_key"},
            "fred": {"enabled": False, "url": "", "name": "FRED Macro Fed (clé requise)", "requires_key": "fred_api_key"},
            "usda": {"enabled": False, "url": "https://apps.fas.usda.gov/psdonline/api/psd/crops", "name": "USDA Agriculture (gratuit)"},
            "who": {"enabled": False, "url": "https://www.who.int/rss-feeds/news-english.xml", "name": "WHO Santé (gratuit)"},
            "emdat": {"enabled": False, "url": "", "name": "EM-DAT Catastrophes (inscription requise)"},
            "twitter_trump": {"enabled": False, "url": "", "name": "X/Twitter Trump (API payante)"},
        }),
        "ai_enabled": "false",
        "ai_auto_rescore": "false",
        "auto_cycle_enabled": "false",
        "auto_cycle_hours": "3",
        "auto_cycle_similarity_threshold": "0.30",
        "min_ev_threshold": "0.0",
        "min_score_threshold": "0",
        "exit_defaults": json.dumps({
            "target_pct": 30.0,
            "stop_loss_pct": -50.0,
            "signal_reversal_mode": "badge_only",
            "signal_reversal_threshold": 25,
        }),
    }
    for k, v in defaults.items():
        c.execute("INSERT OR IGNORE INTO config (key, value) VALUES (?, ?)", (k, v))

    c.execute("""CREATE TABLE IF NOT EXISTS ai_reasoning_traces (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        run_id TEXT NOT NULL,
        trace_type TEXT NOT NULL,
        pattern_id TEXT,
        input_context_json TEXT DEFAULT '{}',
        output_json TEXT DEFAULT '{}',
        reasoning_summary TEXT,
        geo_score INTEGER,
        macro_dominant TEXT,
        created_at TEXT NOT NULL DEFAULT (datetime('now'))
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_art_run ON ai_reasoning_traces(run_id)")
        c.execute("CREATE INDEX IF NOT EXISTS idx_art_pattern ON ai_reasoning_traces(pattern_id, trace_type)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS ai_reports (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        report_type TEXT NOT NULL DEFAULT 'portfolio',
        days INTEGER NOT NULL DEFAULT 30,
        stats_json TEXT DEFAULT '{}',
        winners_json TEXT DEFAULT '[]',
        losers_json TEXT DEFAULT '[]',
        report_json TEXT DEFAULT '{}',
        created_at TEXT NOT NULL DEFAULT (datetime('now'))
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_reports_type_date ON ai_reports(report_type, created_at)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS knowledge_base (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        category TEXT NOT NULL,
        title TEXT NOT NULL,
        content TEXT NOT NULL,
        confidence INTEGER DEFAULT 50,
        confirmation_count INTEGER DEFAULT 1,
        status TEXT DEFAULT 'active',
        tags TEXT DEFAULT '',
        first_seen_at TEXT NOT NULL DEFAULT (datetime('now')),
        last_confirmed_at TEXT NOT NULL DEFAULT (datetime('now'))
    )""")

    c.execute("""CREATE TABLE IF NOT EXISTS reasoning_state (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        version INTEGER NOT NULL DEFAULT 1,
        narrative TEXT NOT NULL,
        synthesis_json TEXT DEFAULT '{}',
        sources_count INTEGER DEFAULT 0,
        reports_used INTEGER DEFAULT 0,
        trades_analyzed INTEGER DEFAULT 0,
        created_at TEXT NOT NULL DEFAULT (datetime('now'))
    )""")
    c.execute("""CREATE TABLE IF NOT EXISTS iv_history (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        ticker TEXT NOT NULL,
        recorded_date TEXT NOT NULL,
        iv_current REAL,
        iv_30d REAL,
        iv_60d REAL,
        iv_90d REAL,
        created_at TEXT DEFAULT (datetime('now'))
    )""")

    c.execute("""CREATE TABLE IF NOT EXISTS iv_watchlist (
        ticker TEXT PRIMARY KEY,
        added_date TEXT NOT NULL DEFAULT (date('now')),
        added_by TEXT DEFAULT 'builtin',
        is_active INTEGER DEFAULT 1
    )""")

    c.execute("""CREATE TABLE IF NOT EXISTS system_logs (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        ts TEXT NOT NULL DEFAULT (datetime('now')),
        level TEXT NOT NULL,
        source TEXT,
        cycle_id TEXT,
        ticker TEXT,
        message TEXT NOT NULL,
        details TEXT
    )""")

    c.execute("""CREATE TABLE IF NOT EXISTS cycle_context_snapshots (
        run_id TEXT PRIMARY KEY,
        ts TEXT NOT NULL DEFAULT (datetime('now')),
        context_json TEXT NOT NULL
    )""")

    c.execute("""CREATE TABLE IF NOT EXISTS ai_call_logs (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        run_id TEXT NOT NULL,
        call_type TEXT NOT NULL,
        pattern_id TEXT,
        pattern_name TEXT,
        system_prompt TEXT,
        user_prompt TEXT,
        response_json TEXT,
        model TEXT DEFAULT 'gpt-4o',
        tokens_prompt INTEGER DEFAULT 0,
        tokens_completion INTEGER DEFAULT 0,
        duration_ms INTEGER DEFAULT 0,
        called_at TEXT NOT NULL DEFAULT (datetime('now'))
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_acl_run ON ai_call_logs(run_id, called_at DESC)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS news_price_snapshots (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        article_hash TEXT NOT NULL,
        article_title TEXT,
        ticker TEXT NOT NULL,
        expected_direction TEXT,
        expected_impact_score REAL,
        price_at_capture REAL,
        captured_at TEXT NOT NULL DEFAULT (datetime('now')),
        capture_cycle_id TEXT,
        UNIQUE(article_hash, ticker)
    )""")

    c.execute("""CREATE TABLE IF NOT EXISTS skipped_trades (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        run_id TEXT,
        pattern_id TEXT,
        pattern_name TEXT,
        underlying TEXT,
        strategy TEXT,
        score INTEGER DEFAULT 0,
        expected_move_pct REAL,
        skip_reason TEXT DEFAULT 'no_profile',
        skip_detail TEXT,
        asset_class TEXT,
        created_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now'))
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_skipped_date ON skipped_trades(created_at DESC)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS var_snapshots (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        computed_at TEXT NOT NULL,
        confidence REAL NOT NULL DEFAULT 0.95,
        horizon_days INTEGER NOT NULL DEFAULT 1,
        lookback_days INTEGER NOT NULL DEFAULT 252,
        default_iv REAL NOT NULL DEFAULT 0.20,
        hist_var_1d_pct REAL,
        hist_cvar_pct REAL,
        hist_var_1d_eur REAL,
        param_var_1d_pct REAL,
        param_cvar_pct REAL,
        mc_var_1d_pct REAL,
        mc_cvar_pct REAL,
        n_positions INTEGER,
        total_notional_eur REAL,
        data_source TEXT,
        breach_rate_pct REAL,
        kupiec_ok INTEGER,
        macro_regime TEXT,
        ticker_prices TEXT,
        full_result TEXT
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_var_snap_ts ON var_snapshots(computed_at DESC)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS pnl_snapshots (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        snapped_at TEXT NOT NULL,
        n_open INTEGER,
        n_closed INTEGER,
        total_capital_eur REAL,
        total_pnl_pct REAL,
        total_pnl_eur REAL,
        ticker_prices TEXT,
        macro_regime TEXT,
        trades_snapshot TEXT
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_pnl_snap_ts ON pnl_snapshots(snapped_at DESC)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS cycle_reports (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        run_id TEXT NOT NULL UNIQUE,
        generated_at TEXT NOT NULL,
        macro_dominant TEXT,
        geo_score INTEGER,
        patterns_added INTEGER DEFAULT 0,
        trades_logged INTEGER DEFAULT 0,
        trades_closed INTEGER DEFAULT 0,
        pnl_snapshot_id INTEGER,
        var_snapshot_id INTEGER,
        full_report_json TEXT
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_cycle_reports_ts ON cycle_reports(generated_at DESC)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS options_trade_assessments (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        run_id TEXT NOT NULL,
        trade_id INTEGER,
        ticker TEXT,
        strategy TEXT,
        assessed_at TEXT,
        iv_rank REAL,
        iv_current_pct REAL,
        skew_pct REAL,
        term_structure TEXT,
        fit_score INTEGER,
        verdict TEXT,
        issues_json TEXT,
        optimal_strategy TEXT,
        analysis TEXT,
        when_to_enter TEXT
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_ota_run ON options_trade_assessments(run_id)")
        c.execute("CREATE INDEX IF NOT EXISTS idx_ota_trade ON options_trade_assessments(trade_id)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS institutional_reports (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        report_type TEXT NOT NULL,
        report_date TEXT NOT NULL,
        fetch_date TEXT NOT NULL DEFAULT (datetime('now')),
        title TEXT NOT NULL,
        source TEXT NOT NULL DEFAULT '',
        importance INTEGER NOT NULL DEFAULT 2,
        category TEXT NOT NULL DEFAULT 'multi',
        raw_data_json TEXT DEFAULT '{}',
        key_points_json TEXT DEFAULT '[]',
        trading_implications TEXT DEFAULT '',
        signal_energy TEXT DEFAULT 'neutral',
        signal_metals TEXT DEFAULT 'neutral',
        signal_indices TEXT DEFAULT 'neutral',
        signal_forex TEXT DEFAULT 'neutral',
        ai_summary TEXT DEFAULT '',
        injected_in_cycle_id TEXT DEFAULT NULL,
        absorbed_score REAL DEFAULT NULL,
        UNIQUE(report_type, report_date)
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_inst_type_date ON institutional_reports(report_type, report_date DESC)")
        c.execute("CREATE INDEX IF NOT EXISTS idx_inst_importance ON institutional_reports(importance DESC, report_date DESC)")
    except Exception:
        pass

    c.execute("""CREATE TABLE IF NOT EXISTS economic_events (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        event_name TEXT NOT NULL,
        series_id TEXT NOT NULL,
        event_date TEXT NOT NULL,
        actual_value REAL,
        actual_unit TEXT DEFAULT '',
        forecast_value REAL,
        previous_value REAL,
        surprise_pct REAL,
        surprise_zscore REAL,
        surprise_direction TEXT DEFAULT 'neutral',
        assets_impacted TEXT DEFAULT '[]',
        source TEXT DEFAULT 'FRED',
        fetched_at TEXT NOT NULL DEFAULT (datetime('now')),
        UNIQUE(series_id, event_date)
    )""")
    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_econ_date ON economic_events(event_date DESC)")
        c.execute("CREATE INDEX IF NOT EXISTS idx_econ_series ON economic_events(series_id, event_date DESC)")
    except Exception:
        pass

    try:
        c.execute("CREATE INDEX IF NOT EXISTS idx_kb_category ON knowledge_base(category, status)")
        c.execute("CREATE INDEX IF NOT EXISTS idx_rs_version ON reasoning_state(version DESC)")
        c.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_iv_history_ticker_date ON iv_history(ticker, recorded_date)")
        c.execute("CREATE INDEX IF NOT EXISTS idx_syslog_ts ON system_logs(ts DESC)")
        c.execute("CREATE INDEX IF NOT EXISTS idx_syslog_level ON system_logs(level, ts DESC)")
        c.execute("CREATE INDEX IF NOT EXISTS idx_syslog_cycle ON system_logs(cycle_id, ts DESC)")
    except Exception:
        pass

    # Seed default config values if not already set
    for _key, _val in [
        ("journal_retention_days", "90"),
        ("maturity_threshold_pct", "35"),
        # IV Gate — blocks ALERT trades before they are logged
        ("iv_gate_enabled", "true"),       # enable/disable the IV gate entirely
        ("iv_gate_ivr_high", "60"),        # IVR above this = "high vol" (no naked long)
        ("iv_gate_ivr_extreme", "80"),     # IVR above this = "extreme vol" (sell vol only)
        ("iv_gate_skew_threshold", "8"),   # put skew above this = protect is expensive
    ]:
        existing = c.execute("SELECT value FROM config WHERE key=?", (_key,)).fetchone()
        if not existing:
            c.execute(
                "INSERT OR IGNORE INTO config (key, value, updated_at) VALUES (?, ?, datetime('now'))",
                (_key, _val)
            )

    # Seed built-in watchlist tickers (idempotent)
    _builtin_tickers = [
        "SPY", "QQQ", "GLD", "SLV", "USO", "BNO", "UNG",
        "XLE", "UUP", "TLT", "GDX", "EWJ", "FEZ",
        "XOM", "CVX", "LMT", "RTX", "BA",
    ]
    for _t in _builtin_tickers:
        c.execute(
            "INSERT OR IGNORE INTO iv_watchlist (ticker, added_by) VALUES (?, 'builtin')",
            (_t,)
        )

    conn.commit()
    conn.close()


# ── Config ────────────────────────────────────────────────────────────────────

def get_config(key: str) -> Optional[str]:
    conn = get_conn()
    row = conn.execute("SELECT value FROM config WHERE key=?", (key,)).fetchone()
    conn.close()
    return row["value"] if row else None


def set_config(key: str, value: str):
    conn = get_conn()
    conn.execute(
        "INSERT OR REPLACE INTO config (key, value, updated_at) VALUES (?, ?, datetime('now'))",
        (key, value)
    )
    conn.commit()
    conn.close()
    if key == "openai_api_key":
        os.environ["OPENAI_API_KEY"] = value


def get_all_config() -> Dict[str, str]:
    conn = get_conn()
    rows = conn.execute("SELECT key, value FROM config").fetchall()
    conn.close()
    result = {r["key"]: r["value"] for r in rows}
    if "openai_api_key" in result and result["openai_api_key"]:
        result["openai_api_key_set"] = True
        result["openai_api_key"] = "***"
    return result


def get_sources() -> Dict[str, Any]:
    raw = get_config("sources")
    return json.loads(raw) if raw else {}


def update_sources(sources: Dict[str, Any]):
    set_config("sources", json.dumps(sources))


def save_pattern_scores(scores: List[Dict[str, Any]], meta: Dict[str, Any] = None) -> str:
    """Persist last AI scores and append a history snapshot. Returns run_id."""
    from datetime import datetime as _dt
    run_id = _dt.utcnow().isoformat()
    data = {"scores": scores, "meta": meta or {}, "scored_at": run_id, "run_id": run_id}
    set_config("last_pattern_scores", json.dumps(data))

    conn = get_conn()
    for sp in scores:
        pid = sp.get("pattern_id", "")
        if pid:
            # Look up predicted_probability from custom_patterns
            row = conn.execute("SELECT probability FROM custom_patterns WHERE id=?", (pid,)).fetchone()
            predicted_prob = float(row["probability"]) if row and row["probability"] is not None else None
            conn.execute(
                "INSERT INTO pattern_score_history (run_id, pattern_id, score, confidence, summary, scored_at, predicted_probability) VALUES (?,?,?,?,?,?,?)",
                (run_id, pid, sp.get("score"), sp.get("confidence"), sp.get("summary", ""), run_id, predicted_prob),
            )
    # Keep only the last 30 runs
    conn.execute("""DELETE FROM pattern_score_history WHERE run_id NOT IN (
        SELECT DISTINCT run_id FROM pattern_score_history ORDER BY scored_at DESC LIMIT 30
    )""")
    conn.commit()
    conn.close()
    return run_id


def get_pattern_scores() -> Dict[str, Any]:
    """Return last persisted AI scores."""
    raw = get_config("last_pattern_scores")
    if raw:
        try:
            return json.loads(raw)
        except Exception:
            pass
    return {"scores": [], "meta": {}, "scored_at": None}


def get_score_deltas() -> Dict[str, int]:
    """Compute score change per pattern between the two most recent scoring runs."""
    conn = get_conn()
    runs = conn.execute(
        "SELECT DISTINCT run_id FROM pattern_score_history ORDER BY scored_at DESC LIMIT 2"
    ).fetchall()
    if len(runs) < 2:
        conn.close()
        return {}
    latest_run, prev_run = runs[0]["run_id"], runs[1]["run_id"]
    latest = {r["pattern_id"]: r["score"] for r in
              conn.execute("SELECT pattern_id, score FROM pattern_score_history WHERE run_id=?", (latest_run,)).fetchall()}
    prev = {r["pattern_id"]: r["score"] for r in
            conn.execute("SELECT pattern_id, score FROM pattern_score_history WHERE run_id=?", (prev_run,)).fetchall()}
    conn.close()
    return {
        pid: score - prev[pid]
        for pid, score in latest.items()
        if pid in prev and score is not None and prev[pid] is not None
    }


def get_score_history(pattern_id: str, limit: int = 10) -> List[Dict[str, Any]]:
    """Return the last N score snapshots for a given pattern."""
    conn = get_conn()
    rows = conn.execute(
        "SELECT score, confidence, summary, scored_at FROM pattern_score_history WHERE pattern_id=? ORDER BY scored_at DESC LIMIT ?",
        (pattern_id, limit),
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def compute_pattern_similarity(patterns: List[Dict[str, Any]], threshold: float = 0.25) -> List[Dict[str, Any]]:
    """Return pairs of patterns with Jaccard keyword similarity above threshold."""
    results = []
    for i, p1 in enumerate(patterns):
        kw1 = set(kw.lower() for kw in (p1.get("keywords") or []))
        for j, p2 in enumerate(patterns):
            if i >= j:
                continue
            kw2 = set(kw.lower() for kw in (p2.get("keywords") or []))
            if not kw1 or not kw2:
                continue
            common = kw1 & kw2
            union = kw1 | kw2
            sim = len(common) / len(union) if union else 0.0
            if sim >= threshold:
                results.append({
                    "id_a": p1.get("id"), "name_a": p1.get("name"),
                    "id_b": p2.get("id"), "name_b": p2.get("name"),
                    "similarity": round(sim, 2),
                    "common_keywords": sorted(common)[:8],
                })
    return sorted(results, key=lambda x: -x["similarity"])


def get_analysis_config() -> Dict[str, Any]:
    """Return the AI analysis config: template, top_n, category_filter."""
    raw = get_config("analysis_config")
    if raw:
        try:
            return json.loads(raw)
        except Exception:
            pass
    return {"top_n": 10, "category_filter": "all", "template": None}


def save_analysis_config(cfg: Dict[str, Any]):
    set_config("analysis_config", json.dumps(cfg))


# ── Portfolio ─────────────────────────────────────────────────────────────────

IB_OPTIONS_FEE_PER_CONTRACT = 0.65
IB_MIN_FEE = 1.0


def compute_ib_fees(num_contracts: int) -> float:
    return max(IB_MIN_FEE, num_contracts * IB_OPTIONS_FEE_PER_CONTRACT)


def add_position(pos: Dict[str, Any]) -> str:
    import uuid
    pos_id = pos.get("id") or f"POS-{uuid.uuid4().hex[:8].upper()}"
    legs = pos.get("legs", [])
    num_contracts = sum(abs(leg.get("quantity", 1)) for leg in legs)
    ib_fees = compute_ib_fees(num_contracts)

    conn = get_conn()
    conn.execute("""INSERT INTO portfolio (
        id, title, underlying, strategy, asset_class, entry_date, expiry_date,
        expiry_days, legs, capital_invested, entry_underlying_price,
        geo_trigger, rationale, status, notes, ib_fees_entry
    ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,'open',?,?)""", (
        pos_id,
        pos.get("title", pos.get("underlying", "")),
        pos.get("underlying", ""),
        pos.get("strategy", ""),
        pos.get("asset_class", ""),
        pos.get("entry_date", datetime.utcnow().isoformat()[:10]),
        pos.get("expiry_date", ""),
        pos.get("expiry_days", 90),
        json.dumps(legs),
        pos.get("capital_invested", 1000.0),
        pos.get("entry_underlying_price"),
        pos.get("geo_trigger", ""),
        pos.get("rationale", ""),
        pos.get("notes", ""),
        ib_fees,
    ))
    conn.commit()
    conn.close()
    return pos_id


def get_positions(status: str = "open") -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM portfolio WHERE status=? ORDER BY created_at DESC",
        (status,)
    ).fetchall()
    conn.close()
    result = []
    for r in rows:
        d = dict(r)
        d["legs"] = json.loads(d.get("legs", "[]"))
        result.append(d)
    return result


def close_position(pos_id: str, close_value: float) -> Dict[str, Any]:
    conn = get_conn()
    pos = conn.execute("SELECT * FROM portfolio WHERE id=?", (pos_id,)).fetchone()
    if not pos:
        conn.close()
        return {"error": "Position non trouvée"}
    legs = json.loads(pos["legs"])
    num_contracts = sum(abs(leg.get("quantity", 1)) for leg in legs)
    ib_exit = compute_ib_fees(num_contracts)
    conn.execute("""UPDATE portfolio SET status='closed', close_date=?, close_value=?,
        ib_fees_exit=? WHERE id=?""",
        (datetime.utcnow().isoformat()[:10], close_value, ib_exit, pos_id))
    conn.commit()
    pnl = close_value - pos["capital_invested"] - pos["ib_fees_entry"] - ib_exit
    conn.close()
    return {"id": pos_id, "close_value": close_value, "ib_fees_exit": ib_exit, "pnl": pnl}


def update_position_notes(pos_id: str, notes: str):
    conn = get_conn()
    conn.execute("UPDATE portfolio SET notes=? WHERE id=?", (notes, pos_id))
    conn.commit()
    conn.close()


# ── Custom Patterns ────────────────────────────────────────────────────────────

def seed_builtin_patterns(builtin_patterns: List[Dict[str, Any]]):
    """Seed built-in patterns into DB (idempotent). Always updates taxonomy_path."""
    conn = get_conn()
    existing = {r[0] for r in conn.execute("SELECT id FROM custom_patterns").fetchall()}
    for p in builtin_patterns:
        taxonomy_json = json.dumps(p.get("taxonomy_path", []))
        if p["id"] not in existing:
            conn.execute("""INSERT INTO custom_patterns (
                id, name, description, triggers, keywords, historical_instances,
                suggested_trades, asset_class, expected_move_pct, probability,
                horizon_days, source, is_active, taxonomy_path, updated_at
            ) VALUES (?,?,?,?,?,?,?,?,?,?,?,'builtin',1,?,datetime('now'))""", (
                p["id"],
                p.get("name", ""),
                p.get("description", ""),
                json.dumps(p.get("triggers", [])),
                json.dumps(p.get("keywords", [])),
                json.dumps(p.get("historical_instances", [])),
                json.dumps(p.get("suggested_trades", [])),
                p.get("asset_class", "indices"),
                p.get("expected_move_pct", 0),
                p.get("probability", 0.5),
                p.get("horizon_days", 30),
                taxonomy_json,
            ))
        else:
            conn.execute(
                "UPDATE custom_patterns SET taxonomy_path=? WHERE id=? AND source='builtin'",
                (taxonomy_json, p["id"])
            )
    conn.commit()
    conn.close()


def save_custom_pattern(pattern: Dict[str, Any]) -> str:
    import uuid
    pat_id = pattern.get("id") or f"P_USER_{uuid.uuid4().hex[:6].upper()}"
    source = pattern.get("source", "custom")
    conn = get_conn()
    conn.execute("""INSERT OR REPLACE INTO custom_patterns (
        id, name, description, triggers, keywords, historical_instances,
        suggested_trades, asset_class, expected_move_pct, probability,
        horizon_days, ai_quality_score, ai_evaluation, source,
        counter_thesis, invalidation_trigger, invalidation_probability,
        category, signal_direction,
        is_active, updated_at
    ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,1,datetime('now'))""", (
        pat_id,
        pattern.get("name", ""),
        pattern.get("description", ""),
        json.dumps(pattern.get("triggers", [])),
        json.dumps(pattern.get("keywords", [])),
        json.dumps(pattern.get("historical_instances", [])),
        json.dumps(pattern.get("suggested_trades", [])),
        pattern.get("asset_class", "indices"),
        pattern.get("expected_move_pct", 0),
        pattern.get("probability", 0.5),
        pattern.get("horizon_days", 30),
        pattern.get("ai_quality_score"),
        json.dumps(pattern.get("ai_evaluation", {})),
        source,
        pattern.get("counter_thesis"),
        pattern.get("invalidation_trigger"),
        pattern.get("invalidation_probability"),
        pattern.get("category"),
        pattern.get("signal_direction"),
    ))
    conn.commit()
    conn.close()
    return pat_id


def get_custom_patterns() -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM custom_patterns WHERE is_active=1 ORDER BY source DESC, created_at DESC"
    ).fetchall()
    conn.close()
    result = []
    for r in rows:
        d = dict(r)
        for f in ["triggers", "keywords", "historical_instances", "suggested_trades", "ai_evaluation"]:
            d[f] = json.loads(d.get(f) or "[]")
        d["taxonomy_path"] = json.loads(d.get("taxonomy_path") or "[]")
        result.append(d)
    return result


def get_unclassified_patterns() -> List[Dict[str, Any]]:
    """Return active patterns where category IS NULL (need AI classification)."""
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM custom_patterns WHERE is_active=1 AND (category IS NULL OR category='')"
    ).fetchall()
    conn.close()
    result = []
    for r in rows:
        d = dict(r)
        for f in ["triggers", "keywords", "historical_instances", "suggested_trades", "ai_evaluation"]:
            d[f] = json.loads(d.get(f) or "[]")
        result.append(d)
    return result


def update_pattern_classification(pat_id: str, category: str, signal_direction: str) -> None:
    """Persist category + signal_direction for a pattern."""
    conn = get_conn()
    conn.execute(
        "UPDATE custom_patterns SET category=?, signal_direction=?, updated_at=datetime('now') WHERE id=?",
        (category, signal_direction, pat_id),
    )
    conn.commit()
    conn.close()


def get_patterns_with_last_score() -> List[Dict[str, Any]]:
    """Each active pattern + its most recent score from pattern_score_history."""
    conn = get_conn()
    rows = conn.execute("""
        SELECT cp.*, psh.score AS last_score, psh.summary AS last_summary, psh.scored_at AS last_scored_at
        FROM custom_patterns cp
        LEFT JOIN (
            SELECT pattern_id, score, summary, scored_at,
                   ROW_NUMBER() OVER (PARTITION BY pattern_id ORDER BY scored_at DESC) AS rn
            FROM pattern_score_history
        ) psh ON psh.pattern_id = cp.id AND psh.rn = 1
        WHERE cp.is_active=1
        ORDER BY cp.created_at DESC
    """).fetchall()
    conn.close()
    result = []
    for r in rows:
        d = dict(r)
        for f in ["triggers", "keywords", "historical_instances", "suggested_trades", "ai_evaluation"]:
            d[f] = json.loads(d.get(f) or "[]")
        result.append(d)
    return result


def toggle_pattern_active(pat_id: str) -> bool:
    """Toggle is_active for a pattern. Returns the new state."""
    conn = get_conn()
    row = conn.execute("SELECT is_active FROM custom_patterns WHERE id=?", (pat_id,)).fetchone()
    if not row:
        conn.close()
        return False
    new_state = 0 if row["is_active"] else 1
    conn.execute("UPDATE custom_patterns SET is_active=? WHERE id=?", (new_state, pat_id))
    conn.commit()
    conn.close()
    return bool(new_state)


def delete_custom_pattern(pat_id: str):
    conn = get_conn()
    conn.execute("UPDATE custom_patterns SET is_active=0 WHERE id=?", (pat_id,))
    conn.commit()
    conn.close()


# ── Risk Profiles ─────────────────────────────────────────────────────────────

def get_risk_profiles(enabled_only: bool = False) -> List[Dict[str, Any]]:
    conn = get_conn()
    q = "SELECT * FROM risk_profiles"
    if enabled_only:
        q += " WHERE enabled=1"
    q += " ORDER BY sort_order ASC, id ASC"
    rows = conn.execute(q).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def upsert_risk_profile(profile: Dict[str, Any]) -> int:
    conn = get_conn()
    pid = profile.get("id")
    if pid:
        conn.execute("""UPDATE risk_profiles
            SET name=?, min_score=?, min_gain_pct=?, color=?, enabled=?, sort_order=?
            WHERE id=?""", (
            profile["name"], int(profile["min_score"]), float(profile["min_gain_pct"]),
            profile.get("color", "#3b82f6"), 1 if profile.get("enabled", True) else 0,
            int(profile.get("sort_order", 0)), pid,
        ))
    else:
        cur = conn.execute("""INSERT INTO risk_profiles (name, min_score, min_gain_pct, color, enabled, sort_order)
            VALUES (?,?,?,?,?,?)""", (
            profile["name"], int(profile["min_score"]), float(profile["min_gain_pct"]),
            profile.get("color", "#3b82f6"), 1 if profile.get("enabled", True) else 0,
            int(profile.get("sort_order", 0)),
        ))
        pid = cur.lastrowid
    conn.commit()
    conn.close()
    return pid


def delete_risk_profile(profile_id: int):
    conn = get_conn()
    conn.execute("DELETE FROM risk_profiles WHERE id=?", (profile_id,))
    conn.commit()
    conn.close()


def _compute_trade_score(score: int, gain_pct: float) -> tuple[float, float, float]:
    """
    Returns (ev_gross, ev_net, trade_score) for a (score, gain_pct) pair.
    - ev_gross = p × G           (raw expected multiple)
    - ev_net   = p × G - (1-p)   (net EV assuming total loss if wrong)
    - trade_score = p×G / (p×G + (1-p)) × 100  (normalized 0-100)
    """
    p = max(0.0, min(1.0, score / 100))
    G = abs(gain_pct) / 100
    ev_gross = round(p * G, 4)
    ev_net = round(p * G - (1 - p), 4)
    denom = p * G + (1 - p)
    trade_score = round((p * G / denom * 100) if denom > 0 else 0.0, 1)
    return ev_gross, ev_net, trade_score


def _matches_profile(score: int, gain_pct: float, profiles: List[Dict[str, Any]]) -> Optional[str]:
    """Return the name of the first enabled profile this trade satisfies, or None."""
    for prof in profiles:
        if not prof.get("enabled", True):
            continue
        if score >= prof["min_score"] and gain_pct >= prof["min_gain_pct"]:
            return prof["name"]
    return None


# ── Journal de Bord ────────────────────────────────────────────────────────────

def log_macro_regime(dominant: str, scores: Dict[str, Any], reasons: Dict[str, Any], gauges_summary: Dict[str, Any]):
    """Append a macro regime snapshot. Keeps last 90 days."""
    conn = get_conn()
    conn.execute("""INSERT INTO macro_regime_history
        (timestamp, dominant, scores_json, reasons_json, gauges_summary_json)
        VALUES (datetime('now'), ?, ?, ?, ?)""",
        (dominant, json.dumps(scores), json.dumps(reasons), json.dumps(gauges_summary)))
    conn.execute("""DELETE FROM macro_regime_history WHERE timestamp < datetime('now', '-90 days')""")
    conn.commit()
    conn.close()


def get_macro_regime_history(days: int = 15) -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM macro_regime_history WHERE timestamp >= datetime('now', ?) ORDER BY timestamp DESC",
        (f"-{days} days",)
    ).fetchall()
    conn.close()
    result = []
    for r in rows:
        d = dict(r)
        d["scores"] = json.loads(d.pop("scores_json", "{}"))
        d["reasons"] = json.loads(d.pop("reasons_json", "{}"))
        d["gauges_summary"] = json.loads(d.pop("gauges_summary_json", "{}"))
        result.append(d)
    return result


def log_geo_alert(geo_score: int, top_patterns: List[Dict[str, Any]], news_count: int, run_id: str):
    """Append a geo alert snapshot tied to a scoring run."""
    conn = get_conn()
    conn.execute("""INSERT INTO geo_alert_history
        (timestamp, run_id, geo_score, top_patterns_json, news_count)
        VALUES (datetime('now'), ?, ?, ?, ?)""",
        (run_id, geo_score, json.dumps(top_patterns[:10]), news_count))
    conn.execute("DELETE FROM geo_alert_history WHERE timestamp < datetime('now', '-90 days')")
    conn.commit()
    conn.close()


def get_geo_alert_history(days: int = 30) -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM geo_alert_history WHERE timestamp >= datetime('now', ?) ORDER BY timestamp DESC",
        (f"-{days} days",)
    ).fetchall()
    conn.close()
    result = []
    for r in rows:
        d = dict(r)
        d["top_patterns"] = json.loads(d.pop("top_patterns_json", "[]"))
        result.append(d)
    return result


def _normalize_yf_ticker(ticker: str) -> str:
    """Normalize ticker for yfinance.
    - USD/KRW  → USDKRW=X  (slash-format forex pairs from GPT-4o)
    - USDKRW   → USDKRW=X  (bare 6-char alphabetic forex pairs)
    - CL=F, SPY, etc. → unchanged
    """
    t = ticker.upper().strip()
    if '/' in t:
        parts = t.split('/')
        if len(parts) == 2 and all(p.isalpha() and len(p) >= 2 for p in parts):
            return parts[0] + parts[1] + '=X'
        return t
    if len(t) == 6 and t.isalpha():
        return t + "=X"
    return t


# Common AI hallucinations / wrong ticker formats → canonical Yahoo Finance symbols
TICKER_ALIASES: dict = {
    # Commodities
    "WHEAT": "ZW=F", "CORN_FUTURES": "ZC=F", "SOYBEANS": "ZS=F", "SOYBEAN": "ZS=F",
    "CRUDE": "CL=F", "OIL": "CL=F", "WTI": "CL=F", "BRENT": "BZ=F",
    "CRUDE OIL": "CL=F", "NATURAL GAS": "NG=F",
    "GOLD": "GC=F", "SILVER": "SI=F", "COPPER": "HG=F", "PLATINUM": "PL=F",
    "SUGAR": "SB=F", "COFFEE": "KC=F", "COCOA": "CC=F", "COTTON": "CT=F",
    # Forex — "/" format → yfinance format
    "EUR/USD": "EURUSD=X", "USD/EUR": "EURUSD=X",
    "USD/JPY": "USDJPY=X", "JPY/USD": "USDJPY=X",
    "GBP/USD": "GBPUSD=X", "USD/GBP": "GBPUSD=X",
    "USD/CHF": "USDCHF=X", "CHF/USD": "USDCHF=X",
    "AUD/USD": "AUDUSD=X", "USD/CAD": "USDCAD=X",
    # Indices
    "SP500": "^GSPC", "S&P500": "^GSPC", "S&P 500": "^GSPC",
    "NASDAQ": "QQQ", "NASDAQ100": "^NDX",
    "DOW": "DIA", "DOW JONES": "DIA",
    "RUSSELL2000": "IWM", "RUSSELL 2000": "IWM",
}


def normalize_ticker(ticker: str) -> str:
    """Normalize AI-generated ticker strings to valid Yahoo Finance symbols."""
    if not ticker:
        return ticker
    t = ticker.strip()
    upper = t.upper()
    # Check alias map (case-insensitive)
    if upper in TICKER_ALIASES:
        return TICKER_ALIASES[upper]
    # Forex: "EUR/USD" style not caught above
    if "/" in t:
        parts = t.upper().split("/")
        if len(parts) == 2 and all(2 <= len(p) <= 4 for p in parts):
            return parts[0] + parts[1] + "=X"
    return t


_TICKER_ASSET_CLASS: dict = {
    # Energy
    "CL=F": "energy", "BZ=F": "energy", "NG=F": "energy", "RB=F": "energy", "HO=F": "energy",
    "XLE": "energy", "XOP": "energy", "USO": "energy", "UCO": "energy", "BOIL": "energy", "UNG": "energy",
    # Metals
    "GC=F": "metals", "SI=F": "metals", "HG=F": "metals", "PA=F": "metals", "PL=F": "metals",
    "GLD": "metals", "IAU": "metals", "SLV": "metals", "GDX": "metals", "GDXJ": "metals",
    "PPLT": "metals", "DBP": "metals", "GLDM": "metals",
    # Agriculture
    "ZC=F": "agriculture", "ZS=F": "agriculture", "ZW=F": "agriculture",
    "CC=F": "agriculture", "KC=F": "agriculture", "CT=F": "agriculture", "OJ=F": "agriculture",
    "CORN": "agriculture", "WEAT": "agriculture", "SOYB": "agriculture", "DBA": "agriculture",
    # Indices
    "^GSPC": "indices", "^NDX": "indices", "^DJI": "indices", "^RUT": "indices",
    "SPY": "indices", "QQQ": "indices", "IWM": "indices", "DIA": "indices", "RSP": "indices",
    "VGK": "indices", "EEM": "indices", "EWZ": "indices", "FXI": "indices",
    "EWJ": "indices", "EFA": "indices", "ACWI": "indices", "INDA": "indices",
    "^NSEI": "indices", "^HSI": "indices", "EWG": "indices", "EWU": "indices", "EWI": "indices",
    # Equities (sector ETFs & individual stocks treated as equities)
    "XLF": "equities", "XLK": "equities", "XLV": "equities", "XLI": "equities",
    "XLP": "equities", "XLU": "equities", "XLY": "equities", "XLRE": "equities",
    "XLB": "equities", "XLC": "equities",
    # Forex
    "DX-Y.NYB": "forex", "UUP": "forex", "FXE": "forex", "FXY": "forex",
    "EUO": "forex", "YCS": "forex", "FXA": "forex", "FXB": "forex", "FXF": "forex",
}


def _asset_class_from_ticker(ticker: str) -> str:
    """Infer asset class from ticker symbol when no explicit class is available."""
    if not ticker:
        return ""
    t = ticker.upper().strip()
    if t in _TICKER_ASSET_CLASS:
        return _TICKER_ASSET_CLASS[t]
    # Futures suffix patterns
    if t.endswith("=F"):
        stem = t[:-2]
        if stem[:2] in ("CL", "RB", "HO", "NG", "BZ"):
            return "energy"
        if stem[:2] in ("GC", "SI", "HG", "PA", "PL"):
            return "metals"
        if stem[:2] in ("ZC", "ZS", "ZW", "CC", "KC", "CT"):
            return "agriculture"
    # Currency pairs
    if "=X" in t or "/" in t:
        return "forex"
    return ""


def _normalize_asset_class(cls: str, ticker: str = "") -> str:
    """Map AI-returned asset_class variants to canonical keys, with ticker fallback."""
    if cls:
        c = cls.lower().strip()
        if any(k in c for k in ("energy", "oil", "gas", "petrol", "brent", "wti")):
            return "energy"
        if any(k in c for k in ("metal", "gold", "silver", "copper", "mining", "precious")):
            return "metals"
        if any(k in c for k in ("agri", "grain", "corn", "wheat", "soy", "crop", "coton", "coffee", "cocoa")):
            return "agriculture"
        if any(k in c for k in ("index", "indic", "indices", "spx", "nasdaq", "dow", "s&p", "russell", "cac", "dax")):
            return "indices"
        if any(k in c for k in ("equit", "stock", "action", "share", "sector")):
            return "equities"
        if any(k in c for k in ("forex", "currency", "fx", "devise", "change", "eur", "usd", "jpy", "dxy")):
            return "forex"
        # Already a canonical value
        if c in ("energy", "metals", "agriculture", "indices", "equities", "forex"):
            return c
    # Fallback: infer from ticker
    return _asset_class_from_ticker(ticker)


def log_trade_entries(run_id: str, scored_patterns: List[Dict[str, Any]], quotes: Dict[str, Any]):
    """
    For each scored pattern's trade_rankings, record entry price if the trade
    passes at least one enabled risk profile (min_score + min_gain_pct pair).
    Deduplicates: one row per (pattern_id, underlying, strategy).
    Falls back to yfinance for tickers not found in the quotes snapshot.
    """
    import logging as _logging
    _log = _logging.getLogger(__name__)
    profiles = get_risk_profiles(enabled_only=True)
    _log.info(f"[TradeLog] run_id={run_id} scored_patterns={len(scored_patterns)} profiles={len(profiles)}")

    # Load original patterns as fallback for expected_move_pct
    # (GPT-4o scored output doesn't include this field)
    _orig_patterns = {p.get("id", ""): p for p in get_custom_patterns()}

    # Build price map from pre-fetched quotes
    price_map: Dict[str, float] = {}
    for asset_class, items in quotes.items():
        if isinstance(items, list):
            for item in items:
                if item.get("ticker") and item.get("price") is not None:
                    price_map[item["ticker"].upper()] = float(item["price"])
        elif isinstance(items, dict):
            for ticker, item in items.items():
                if isinstance(item, dict) and item.get("price") is not None:
                    price_map[ticker.upper()] = float(item["price"])

    # Collect tickers NOT already in price_map for yfinance fallback
    tickers_to_fetch: set = set()
    for sp in scored_patterns:
        for trade in sp.get("trade_rankings") or sp.get("suggested_trades", []):
            t = (trade.get("underlying") or trade.get("ticker", "")).upper()
            if t and t not in price_map:
                tickers_to_fetch.add(t)

    if tickers_to_fetch:
        _log.info(f"[TradeLog] yfinance fallback for {len(tickers_to_fetch)} tickers: {sorted(tickers_to_fetch)}")
        try:
            import yfinance as yf
            import pandas as pd
            from concurrent.futures import ThreadPoolExecutor, as_completed

            def _fetch(ticker: str):
                normalized = _normalize_ticker(ticker)  # handles WHEAT→ZW=F, EUR/USD→EURUSD=X, etc.
                try:
                    # Use yf.download() — fresh HTTP request, no in-process Ticker cache
                    for kwargs in [
                        {"period": "1d", "interval": "5m"},
                        {"period": "5d", "interval": "1d"},
                    ]:
                        df = yf.download(normalized, progress=False, auto_adjust=True, **kwargs)
                        if df.empty:
                            continue
                        if isinstance(df.columns, pd.MultiIndex):
                            df.columns = df.columns.get_level_values(0)
                        if "Close" not in df.columns:
                            continue
                        close = df["Close"].dropna()
                        if close.empty:
                            continue
                        price = float(close.iloc[-1])
                        if price > 0:
                            _log.debug(f"[TradeLog] {ticker} → {normalized} = {price}")
                            return ticker, price
                except Exception as e:
                    _log.warning(f"[TradeLog] Failed to fetch '{ticker}' (normalized='{normalized}'): {e}")
                return ticker, None

            with ThreadPoolExecutor(max_workers=min(len(tickers_to_fetch), 10)) as ex:
                for fut in as_completed({ex.submit(_fetch, t): t for t in tickers_to_fetch}, timeout=20):
                    try:
                        tk, price = fut.result()
                        if price is not None:
                            price_map[tk] = price
                    except Exception:
                        pass
        except Exception as e:
            _log.error(f"[TradeLog] yfinance fallback failed: {e}")

    conn = get_conn()
    today = datetime.utcnow().isoformat()[:10]
    now_ts = datetime.utcnow().isoformat()

    inserted_count = 0
    updated_count = 0
    skipped_no_profile = 0

    for sp in scored_patterns:
        pid = sp.get("pattern_id", "")
        pattern_name = sp.get("geo_trigger") or sp.get("pattern_name") or pid
        base_score = int(sp.get("score") or 0)
        _orig = _orig_patterns.get(pid, {})

        for trade in sp.get("trade_rankings") or sp.get("suggested_trades", []):
            underlying = trade.get("underlying") or trade.get("ticker", "")
            if not underlying:
                continue
            strategy = trade.get("strategy") or trade.get("trade_type", "")

            delta = int(trade.get("score_delta") or 0)
            eff_score = max(0, min(100, base_score + delta))
            # Fallback chain: trade field → scored sp field → auto_cycle enrichment → original DB pattern
            exp_move = abs(float(
                trade.get("expected_move_pct") or
                sp.get("expected_move_pct") or
                _orig.get("expected_move_pct") or
                0
            ))
            if exp_move == 0:
                _log.warning(f"[TradeLog] Pattern '{pattern_name}' trade {underlying} has expected_move_pct=0 — all profiles with min_gain_pct>0 will fail")

            # Check if this trade passes any enabled risk profile
            matched = _matches_profile(eff_score, exp_move, profiles)
            if matched is None:
                skipped_no_profile += 1
                _log.debug(f"[TradeLog] SKIP {underlying} score={eff_score} gain={exp_move:.0f}% — no profile match")
                _trade_ac = trade.get("asset_class") or sp.get("asset_class") or _orig.get("asset_class") or ""
                try:
                    log_skipped_trade(
                        run_id=run_id, pattern_id=pid, pattern_name=pattern_name,
                        underlying=underlying, strategy=strategy, score=eff_score,
                        expected_move_pct=exp_move,
                        skip_detail=f"profiles checked: {len(profiles)}, best: score>={eff_score} gain>={exp_move:.0f}%",
                        asset_class=_trade_ac,
                    )
                except Exception:
                    pass
                continue

            ev_gross, ev_net, trade_score = _compute_trade_score(eff_score, exp_move)

            ticker_key = _normalize_ticker(underlying.upper())
            entry_price = price_map.get(ticker_key)
            horizon = int(
                trade.get("horizon_days") or
                sp.get("horizon_days") or
                sp.get("recommended_trade", {}).get("expiry_days") or
                _orig.get("horizon_days") or
                90
            )
            strike_guidance = (
                trade.get("strike_guidance") or
                sp.get("recommended_trade", {}).get("strike_guidance") or
                None
            )
            expiry_days_entry = int(
                trade.get("expiry_days") or
                sp.get("recommended_trade", {}).get("expiry_days") or
                horizon
            )

            existing_row = conn.execute(
                "SELECT id FROM trade_entry_prices WHERE pattern_id=? AND underlying=? AND strategy=?",
                (pid, ticker_key, strategy)
            ).fetchone()

            if existing_row:
                conn.execute(
                    "UPDATE trade_entry_prices SET latest_score=?, trade_score=?, last_seen_at=? WHERE id=?",
                    (eff_score, trade_score, now_ts, existing_row["id"])
                )
                updated_count += 1
            else:
                _trade_asset_class = _normalize_asset_class(
                    trade.get("asset_class") or
                    sp.get("asset_class") or
                    _orig.get("asset_class") or
                    "",
                    ticker=ticker_key
                )
                conn.execute("""INSERT INTO trade_entry_prices
                    (run_id, pattern_id, pattern_name, underlying, strategy,
                     entry_price, entry_date, score_at_entry, latest_score,
                     expected_move_pct, horizon_days, ev_at_entry, ev_net,
                     trade_score, matched_profile, last_seen_at,
                     strike_guidance, expiry_days_at_entry, asset_class)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", (
                    run_id, pid, pattern_name, ticker_key, strategy,
                    entry_price, today, eff_score, eff_score,
                    exp_move, horizon, ev_gross, ev_net,
                    trade_score, matched, now_ts,
                    strike_guidance, expiry_days_entry, _trade_asset_class,
                ))
                inserted_count += 1
                _log.info(f"[TradeLog] NEW trade: pattern='{pattern_name}' {underlying} {strategy} score={eff_score} gain={exp_move:.0f}% profile='{matched}' price={entry_price}")

    _log.info(f"[TradeLog] Done — inserted={inserted_count} updated={updated_count} skipped_no_profile={skipped_no_profile}")
    _retention = int(get_config("journal_retention_days") or "90")
    conn.execute(f"DELETE FROM trade_entry_prices WHERE entry_date < date('now', '-{_retention} days')")
    conn.commit()
    conn.close()


def reset_journal_history():
    """Truncate all journal history tables for a clean slate."""
    conn = get_conn()
    conn.execute("DELETE FROM trade_entry_prices")
    conn.execute("DELETE FROM macro_regime_history")
    conn.execute("DELETE FROM geo_alert_history")
    conn.execute("DELETE FROM cycle_runs")
    conn.commit()
    conn.close()


def get_last_completed_cycle_ts() -> Optional[str]:
    """Return ISO timestamp of the last successfully completed cycle, or None."""
    conn = get_conn()
    row = conn.execute(
        "SELECT completed_at FROM cycle_runs WHERE status='completed' ORDER BY completed_at DESC LIMIT 1"
    ).fetchone()
    conn.close()
    return row["completed_at"] if row else None


def add_cycle_run(run_id: str, trigger: str = "auto") -> None:
    conn = get_conn()
    conn.execute(
        "INSERT OR IGNORE INTO cycle_runs (run_id, started_at, trigger, status) VALUES (?, datetime('now'), ?, 'running')",
        (run_id, trigger)
    )
    conn.commit()
    conn.close()


def cleanup_stale_running_cycles() -> int:
    """Mark any cycle_runs still in 'running' state as 'error' (stale from a crashed process)."""
    conn = get_conn()
    cur = conn.execute(
        "UPDATE cycle_runs SET status='error', completed_at=datetime('now') WHERE status='running'"
    )
    conn.commit()
    conn.close()
    return cur.rowcount


def update_cycle_run(run_id: str, **fields) -> None:
    if not fields:
        return
    allowed = {"completed_at", "patterns_suggested", "patterns_added", "patterns_scored",
               "geo_score", "dominant_regime", "commentary", "status"}
    sets = ", ".join(f"{k}=?" for k in fields if k in allowed)
    vals = [v for k, v in fields.items() if k in allowed]
    if not sets:
        return
    conn = get_conn()
    conn.execute(f"UPDATE cycle_runs SET {sets} WHERE run_id=?", vals + [run_id])
    conn.commit()
    conn.close()


def get_cycle_runs(limit: int = 30) -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM cycle_runs ORDER BY started_at DESC LIMIT ?", (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_cycle_run(run_id: str) -> Optional[Dict[str, Any]]:
    conn = get_conn()
    row = conn.execute("SELECT * FROM cycle_runs WHERE run_id=?", (run_id,)).fetchone()
    conn.close()
    return dict(row) if row else None


def _trade_maturity(days_held: int, horizon_days: int) -> Dict[str, Any]:
    """
    Classify a trade's maturity based on elapsed time vs planned horizon.
    Thresholds read from config (maturity_threshold_pct, default 35%).
    """
    h = max(horizon_days or 90, 1)
    d = max(days_held or 0, 0)
    ratio = d / h
    pct = round(ratio * 100, 1)
    mature_threshold = float(get_config("maturity_threshold_pct") or "35") / 100.0

    if ratio < 0.10:
        return {
            "status": "trop_tot", "label": "Trop tôt", "emoji": "🕐",
            "weight": 0.0, "color": "slate", "ratio_pct": pct,
            "readable": f"{d}j / {h}j ({pct}% écoulé — bruit statistique)",
        }
    elif ratio < mature_threshold:
        return {
            "status": "debut", "label": "Début", "emoji": "📊",
            "weight": 0.25, "color": "yellow", "ratio_pct": pct,
            "readable": f"{d}j / {h}j ({pct}% écoulé — signal précoce)",
        }
    elif ratio < 0.75:
        return {
            "status": "mature", "label": "Signal fiable", "emoji": "✅",
            "weight": 1.0, "color": "emerald", "ratio_pct": pct,
            "readable": f"{d}j / {h}j ({pct}% écoulé — signal fiable)",
        }
    else:
        return {
            "status": "fin_horizon", "label": "Fin d'horizon", "emoji": "⏰",
            "weight": 1.0, "color": "orange", "ratio_pct": pct,
            "readable": f"{d}j / {h}j ({pct}% écoulé — surveiller de près)",
        }


def get_trade_entry_prices(days: int = 30) -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        """SELECT * FROM trade_entry_prices
           WHERE (status IS NULL OR status = 'open')
             AND entry_date >= date('now', ?)
           ORDER BY entry_date DESC, score_at_entry DESC""",
        (f"-{days} days",)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_closed_trades(days: int = 180) -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        """SELECT * FROM trade_entry_prices
           WHERE status = 'closed'
             AND closed_at >= date('now', ?)
           ORDER BY closed_at DESC""",
        (f"-{days} days",)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def delete_trade(trade_id: int) -> bool:
    conn = get_conn()
    cur = conn.execute("DELETE FROM trade_entry_prices WHERE id = ?", (trade_id,))
    conn.commit()
    conn.close()
    return cur.rowcount > 0


def get_skipped_trades(days: int = 30) -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        """SELECT * FROM skipped_trades
           WHERE created_at >= date('now', ?)
           ORDER BY created_at DESC, score DESC""",
        (f"-{days} days",)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def log_skipped_trade(run_id: str, pattern_id: str, pattern_name: str,
                      underlying: str, strategy: str, score: int,
                      expected_move_pct: float, skip_reason: str = "no_profile",
                      skip_detail: str = "", asset_class: str = "") -> None:
    conn = get_conn()
    conn.execute(
        """INSERT INTO skipped_trades
           (run_id, pattern_id, pattern_name, underlying, strategy, score,
            expected_move_pct, skip_reason, skip_detail, asset_class)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (run_id, pattern_id, pattern_name, underlying, strategy, score,
         expected_move_pct, skip_reason, skip_detail, _normalize_asset_class(asset_class, ticker=underlying))
    )
    conn.commit()
    conn.close()


def close_trade(trade_id: int, close_price: float, pnl_realized: float,
                close_reason: str, close_note: str = "") -> bool:
    conn = get_conn()
    cur = conn.execute(
        """UPDATE trade_entry_prices
           SET status='closed', closed_at=datetime('now'), close_price=?,
               pnl_realized=?, close_reason=?, close_note=?
           WHERE id=? AND (status IS NULL OR status='open')""",
        (close_price, pnl_realized, close_reason, close_note, trade_id)
    )
    conn.commit()
    conn.close()
    return cur.rowcount > 0


def update_trade_exit_params(trade_id: int, target_pct: float = None,
                             stop_loss_pct: float = None,
                             signal_threshold: float = None) -> bool:
    updates: List[str] = []
    vals: List[Any] = []
    if target_pct is not None:
        updates.append("target_pct=?"); vals.append(target_pct)
    if stop_loss_pct is not None:
        updates.append("stop_loss_pct=?"); vals.append(stop_loss_pct)
    if signal_threshold is not None:
        updates.append("signal_threshold=?"); vals.append(signal_threshold)
    if not updates:
        return False
    conn = get_conn()
    cur = conn.execute(
        f"UPDATE trade_entry_prices SET {', '.join(updates)} WHERE id=?",
        vals + [trade_id]
    )
    conn.commit()
    conn.close()
    return cur.rowcount > 0


def get_trade_entry_by_id(trade_id: int) -> Optional[Dict[str, Any]]:
    conn = get_conn()
    row = conn.execute("SELECT * FROM trade_entry_prices WHERE id=?", (trade_id,)).fetchone()
    conn.close()
    return dict(row) if row else None


# ── AI Reasoning Traces ───────────────────────────────────────────────────────

def save_reasoning_trace(
    run_id: str,
    trace_type: str,
    pattern_id: str = None,
    input_context: Dict[str, Any] = None,
    output: Dict[str, Any] = None,
    reasoning_summary: str = None,
    geo_score: int = None,
    macro_dominant: str = None,
) -> int:
    """Persist one AI reasoning step. Returns the new row id."""
    conn = get_conn()
    cur = conn.execute(
        """INSERT INTO ai_reasoning_traces
           (run_id, trace_type, pattern_id, input_context_json, output_json,
            reasoning_summary, geo_score, macro_dominant)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
        (
            run_id, trace_type, pattern_id,
            json.dumps(input_context or {}, ensure_ascii=False, default=str),
            json.dumps(output or {}, ensure_ascii=False, default=str),
            reasoning_summary, geo_score, macro_dominant,
        ),
    )
    row_id = cur.lastrowid
    conn.commit()
    conn.close()
    return row_id


def _parse_trace(row) -> Dict[str, Any]:
    """Deserialize a reasoning trace row."""
    d = dict(row)
    for field in ("input_context_json", "output_json"):
        try:
            d[field.replace("_json", "")] = json.loads(d.get(field) or "{}")
        except Exception:
            d[field.replace("_json", "")] = {}
    return d


def get_scoring_trace(run_id: str, pattern_id: str) -> Optional[Dict[str, Any]]:
    """Return the scoring trace for a given (run_id, pattern_id) pair."""
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM ai_reasoning_traces WHERE run_id=? AND pattern_id=? AND trace_type='scoring'",
        (run_id, pattern_id),
    ).fetchone()
    conn.close()
    return _parse_trace(row) if row else None


def get_suggestion_trace(pattern_id: str) -> Optional[Dict[str, Any]]:
    """Return the original suggestion trace for a pattern (first one ever)."""
    conn = get_conn()
    row = conn.execute(
        """SELECT * FROM ai_reasoning_traces
           WHERE pattern_id=? AND trace_type='suggestion'
           ORDER BY created_at ASC LIMIT 1""",
        (pattern_id,),
    ).fetchone()
    conn.close()
    return _parse_trace(row) if row else None


def get_pattern_scoring_history(pattern_id: str, limit: int = 10) -> List[Dict[str, Any]]:
    """All scoring traces for a pattern across cycles — for trend analysis."""
    conn = get_conn()
    rows = conn.execute(
        """SELECT * FROM ai_reasoning_traces
           WHERE pattern_id=? AND trace_type='scoring'
           ORDER BY created_at DESC LIMIT ?""",
        (pattern_id, limit),
    ).fetchall()
    conn.close()
    return [_parse_trace(r) for r in rows]


_BEARISH_KEYWORDS = {"bear", "put", "short", "sell", "vente", "baissier"}


_EXCHANGE_PREFIX_MAP = {
    "NSE": ".NS", "BSE": ".BO", "TSX": ".TO", "LSE": ".L",
    "HKG": ".HK", "SHA": ".SS", "SHE": ".SZ", "TYO": ".T",
}

# Commodity names that GPT-4o sometimes suggests as tickers
_COMMODITY_ALIAS: Dict[str, str] = {
    "WHEAT": "ZW=F", "CORN": "ZC=F", "SOYBEANS": "ZS=F", "SOYBEAN": "ZS=F",
    "SUGAR": "SB=F", "COFFEE": "KC=F", "COTTON": "CT=F",
    "OIL": "CL=F", "CRUDE": "CL=F", "CRUDEOIL": "CL=F", "WTI": "CL=F",
    "BRENT": "BZ=F", "NATGAS": "NG=F", "GAS": "NG=F",
    "GOLD": "GC=F", "SILVER": "SI=F", "COPPER": "HG=F",
    "PLATINUM": "PL=F", "PALLADIUM": "PA=F",
    "NASDAQ": "QQQ", "SP500": "SPY", "DOW": "DIA",
    "VIX": "^VIX", "DOLLAR": "DX-Y.NYB", "DXY": "DX-Y.NYB",
    "EURUSD": "EURUSD=X", "GBPUSD": "GBPUSD=X", "USDJPY": "USDJPY=X",
    "USDCHF": "USDCHF=X", "AUDUSD": "AUDUSD=X", "USDCAD": "USDCAD=X",
}


def _normalize_ticker(ticker: str) -> str:
    """Convert various ticker formats (from GPT-4o) to yfinance-compatible tickers."""
    t = ticker.strip()
    if not t:
        return t
    # exchange:symbol format (e.g. NSE:RELIANCE → RELIANCE.NS)
    if ":" in t:
        exchange, symbol = t.split(":", 1)
        suffix = _EXCHANGE_PREFIX_MAP.get(exchange.upper(), "")
        return symbol + suffix if suffix else symbol
    # slash forex format (e.g. EUR/USD → EURUSD=X)
    if "/" in t:
        parts = t.upper().split("/")
        if len(parts) == 2 and all(p.isalpha() and len(p) >= 2 for p in parts):
            return parts[0] + parts[1] + "=X"
        return t
    u = t.upper()
    # commodity / index aliases (checked first so EURUSD etc. in table win)
    if u in _COMMODITY_ALIAS:
        return _COMMODITY_ALIAS[u]
    # bare 6-char alphabetic forex pairs not already in alias table (e.g. USDSEK → USDSEK=X)
    if len(u) == 6 and u.isalpha():
        _MAJORS = {"USD", "EUR", "GBP", "JPY", "CHF", "AUD", "CAD", "NZD", "SEK", "NOK"}
        if u[3:] in _MAJORS or u[:3] in _MAJORS:
            return u + "=X"
    return t


def _fetch_live_prices(tickers: List[str], timeout: int = 20) -> Dict[str, Optional[float]]:
    """
    Fetch current prices for a list of tickers using yf.download().
    Shared by journal MTM and portfolio report so both get consistent live data.
    """
    result: Dict[str, Optional[float]] = {t: None for t in tickers}
    if not tickers:
        return result
    try:
        import yfinance as yf
        import pandas as pd
        from concurrent.futures import ThreadPoolExecutor, as_completed

        def _one(ticker: str) -> tuple:
            yf_ticker = _normalize_ticker(ticker)
            for kwargs in [
                {"period": "1d", "interval": "5m"},
                {"period": "5d", "interval": "1d"},
            ]:
                try:
                    df = yf.download(yf_ticker, progress=False, auto_adjust=True, **kwargs)
                    if df.empty:
                        continue
                    if isinstance(df.columns, pd.MultiIndex):
                        df.columns = df.columns.get_level_values(0)
                    if "Close" not in df.columns:
                        continue
                    close = df["Close"].dropna()
                    if close.empty:
                        continue
                    price = float(close.iloc[-1])
                    if price > 0:
                        return ticker, price
                except Exception:
                    continue
            return ticker, None

        with ThreadPoolExecutor(max_workers=min(len(tickers), 10)) as ex:
            futs = {ex.submit(_one, t): t for t in tickers}
            from concurrent.futures import as_completed as _ac
            for fut in _ac(futs, timeout=timeout):
                try:
                    tk, price = fut.result()
                    result[tk] = price
                except Exception:
                    pass
    except Exception:
        pass
    return result


def get_mtm_trades_with_traces(days: int = 90, limit_movers: int = 5) -> Dict[str, Any]:
    """
    Return all MTM trades with live prices and reasoning traces for the
    top winners and losers (by pnl_pct). Used by the AI portfolio report.
    """
    from datetime import timedelta
    cutoff_date = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%d")

    conn = get_conn()
    rows = conn.execute(
        """SELECT * FROM trade_entry_prices
           WHERE entry_date >= ?
           ORDER BY entry_date DESC""",
        (cutoff_date,),
    ).fetchall()
    conn.close()

    all_trades = [dict(r) for r in rows]

    # Fetch live prices for all unique tickers
    tickers = list({(t.get("underlying") or "").upper() for t in all_trades if t.get("underlying")})
    live_prices = _fetch_live_prices(tickers, timeout=25)

    # Enrich trades with live price + pnl_pct
    def _with_pnl(trade: Dict) -> Dict:
        ticker = (trade.get("underlying") or "").upper()
        entry = trade.get("entry_price")
        current = live_prices.get(ticker)
        pnl_pct = None
        if entry and current and entry > 0:
            raw = (current - entry) / entry * 100
            strategy = trade.get("strategy", "").lower()
            bearish = any(k in strategy for k in _BEARISH_KEYWORDS)
            pnl_pct = round(-raw if bearish else raw, 2)
        return {**trade, "current_price": current, "pnl_pct": pnl_pct}

    enriched = [_with_pnl(t) for t in all_trades]
    priced = [t for t in enriched if t.get("pnl_pct") is not None]
    winners = sorted(priced, key=lambda t: t.get("pnl_pct", 0), reverse=True)[:limit_movers]
    losers  = sorted(priced, key=lambda t: t.get("pnl_pct", 0))[:limit_movers]

    def _with_traces(trade: Dict) -> Dict:
        pid = trade.get("pattern_id", "")
        run_id = trade.get("run_id", "")
        sc = get_scoring_trace(run_id, pid) if run_id and pid else None
        sg = get_suggestion_trace(pid) if pid else None
        history = get_pattern_scoring_history(pid, limit=5) if pid else []
        return {
            **trade,
            "scoring_context": sc,
            "suggestion_context": sg,
            "score_history_count": len(history),
            "score_trend": [h["output"].get("score") for h in reversed(history)] if history else [],
        }

    return {
        "total_trades": len(all_trades),
        "priced_count": len(priced),
        "avg_pnl_pct": (sum(t.get("pnl_pct", 0) for t in priced) / len(priced)) if priced else None,
        "winners": [_with_traces(t) for t in winners],
        "losers":  [_with_traces(t) for t in losers],
        "all_trades": enriched,
    }


def save_ai_report(
    days: int,
    stats: Dict[str, Any],
    winners: List[Dict],
    losers: List[Dict],
    report: Dict[str, Any],
    report_type: str = "portfolio",
) -> int:
    conn = get_conn()
    cur = conn.execute(
        """INSERT INTO ai_reports
           (report_type, days, stats_json, winners_json, losers_json, report_json)
           VALUES (?, ?, ?, ?, ?, ?)""",
        (
            report_type, days,
            json.dumps(stats, ensure_ascii=False, default=str),
            json.dumps(winners, ensure_ascii=False, default=str),
            json.dumps(losers, ensure_ascii=False, default=str),
            json.dumps(report, ensure_ascii=False, default=str),
        ),
    )
    row_id = cur.lastrowid
    conn.commit()
    conn.close()
    return row_id


def _parse_report(row) -> Dict[str, Any]:
    d = dict(row)
    for field in ("stats_json", "winners_json", "losers_json", "report_json"):
        key = field.replace("_json", "")
        try:
            d[key] = json.loads(d.get(field) or "{}")
        except Exception:
            d[key] = {}
    return d


def get_latest_portfolio_lessons() -> Optional[Dict[str, Any]]:
    """
    Return the key lessons from the most recent portfolio report.
    Used by auto_cycle to inject feedback into the next AI cycle's prompts.
    Returns None if no report exists yet.
    """
    conn = get_conn()
    row = conn.execute(
        """SELECT report_json, stats_json, created_at, days
           FROM ai_reports WHERE report_type='portfolio'
           ORDER BY created_at DESC LIMIT 1"""
    ).fetchone()
    conn.close()
    if not row:
        return None
    try:
        report = json.loads(row["report_json"] or "{}")
        stats  = json.loads(row["stats_json"]  or "{}")
    except Exception:
        return None
    if not report:
        return None
    return {
        "created_at": row["created_at"],
        "days": row["days"],
        "stats": stats,
        "headline": report.get("headline", ""),
        "winners_analysis": report.get("winners_analysis", ""),
        "losers_analysis": report.get("losers_analysis", ""),
        "key_lessons": report.get("key_lessons", []),
        "blind_spots": report.get("blind_spots", ""),
        "next_cycle_priorities": report.get("next_cycle_priorities", ""),
        "risk_watch": report.get("risk_watch", ""),
    }


def list_ai_reports(report_type: str = "portfolio", limit: int = 20) -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        """SELECT id, report_type, days, stats_json, report_json, created_at
           FROM ai_reports
           WHERE report_type=?
           ORDER BY created_at DESC LIMIT ?""",
        (report_type, limit),
    ).fetchall()
    conn.close()
    result = []
    for row in rows:
        d = dict(row)
        for field in ("stats_json", "report_json"):
            key = field.replace("_json", "")
            try:
                d[key] = json.loads(d.get(field) or "{}")
            except Exception:
                d[key] = {}
        result.append(d)
    return result


def get_ai_report(report_id: int) -> Optional[Dict[str, Any]]:
    conn = get_conn()
    row = conn.execute("SELECT * FROM ai_reports WHERE id=?", (report_id,)).fetchone()
    conn.close()
    return _parse_report(row) if row else None


# ── Knowledge Base ─────────────────────────────────────────────────────────────

def save_kb_entry(category: str, title: str, content: str, confidence: int = 50,
                  tags: str = "", existing_id: Optional[int] = None) -> int:
    conn = get_conn()
    now = datetime.utcnow().isoformat()
    if existing_id:
        conn.execute("""UPDATE knowledge_base SET content=?, confidence=?, tags=?,
            last_confirmed_at=?, confirmation_count=confirmation_count+1
            WHERE id=?""", (content, confidence, tags, now, existing_id))
        conn.commit()
        conn.close()
        return existing_id
    cur = conn.execute("""INSERT INTO knowledge_base
        (category, title, content, confidence, confirmation_count, status, tags, first_seen_at, last_confirmed_at)
        VALUES (?,?,?,?,1,'active',?,?,?)""",
        (category, title, content, confidence, tags, now, now))
    new_id = cur.lastrowid
    conn.commit()
    conn.close()
    return new_id


def get_kb_entries(status: str = "active") -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM knowledge_base WHERE status=? ORDER BY confidence DESC, last_confirmed_at DESC",
        (status,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_all_kb_entries() -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM knowledge_base ORDER BY confidence DESC, last_confirmed_at DESC"
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def update_kb_entry_status(entry_id: int, status: str):
    conn = get_conn()
    conn.execute("UPDATE knowledge_base SET status=? WHERE id=?", (status, entry_id))
    conn.commit()
    conn.close()


# ── Reasoning State ────────────────────────────────────────────────────────────

def save_reasoning_state(narrative: str, synthesis: Dict[str, Any],
                         sources_count: int = 0, reports_used: int = 0,
                         trades_analyzed: int = 0) -> int:
    conn = get_conn()
    cur_row = conn.execute(
        "SELECT COALESCE(MAX(version), 0) as v FROM reasoning_state"
    ).fetchone()
    next_version = (cur_row["v"] if cur_row else 0) + 1
    cur = conn.execute("""INSERT INTO reasoning_state
        (version, narrative, synthesis_json, sources_count, reports_used, trades_analyzed, created_at)
        VALUES (?,?,?,?,?,?,datetime('now'))""",
        (next_version, narrative, json.dumps(synthesis), sources_count, reports_used, trades_analyzed))
    new_id = cur.lastrowid
    conn.commit()
    conn.close()
    return new_id


def get_latest_reasoning_state() -> Optional[Dict[str, Any]]:
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM reasoning_state ORDER BY version DESC LIMIT 1"
    ).fetchone()
    conn.close()
    if not row:
        return None
    d = dict(row)
    try:
        d["synthesis"] = json.loads(d.get("synthesis_json") or "{}")
    except Exception:
        d["synthesis"] = {}
    return d


def get_reasoning_history(limit: int = 10) -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT id, version, sources_count, reports_used, trades_analyzed, created_at FROM reasoning_state ORDER BY version DESC LIMIT ?",
        (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_reasoning_state_by_id(state_id: int) -> Optional[Dict[str, Any]]:
    conn = get_conn()
    row = conn.execute("SELECT * FROM reasoning_state WHERE id=?", (state_id,)).fetchone()
    conn.close()
    if not row:
        return None
    d = dict(row)
    try:
        d["synthesis"] = json.loads(d.get("synthesis_json") or "{}")
    except Exception:
        d["synthesis"] = {}
    return d


def delete_ai_report(report_id: int) -> bool:
    conn = get_conn()
    cur = conn.execute("DELETE FROM ai_reports WHERE id=?", (report_id,))
    conn.commit()
    conn.close()
    return cur.rowcount > 0


def delete_reasoning_state(state_id: int) -> bool:
    conn = get_conn()
    cur = conn.execute("DELETE FROM reasoning_state WHERE id=?", (state_id,))
    conn.commit()
    conn.close()
    return cur.rowcount > 0


def delete_kb_entry(entry_id: int) -> bool:
    conn = get_conn()
    cur = conn.execute("DELETE FROM knowledge_base WHERE id=?", (entry_id,))
    conn.commit()
    conn.close()
    return cur.rowcount > 0


# ── IV History ────────────────────────────────────────────────────────────────

def save_iv_snapshot(ticker: str, recorded_date: str, iv_current: float,
                     iv_30d=None, iv_60d=None, iv_90d=None) -> None:
    conn = get_conn()
    conn.execute(
        """INSERT OR REPLACE INTO iv_history
           (ticker, recorded_date, iv_current, iv_30d, iv_60d, iv_90d)
           VALUES (?, ?, ?, ?, ?, ?)""",
        (ticker.upper(), recorded_date, iv_current, iv_30d, iv_60d, iv_90d),
    )
    conn.commit()
    conn.close()


def get_iv_rank_percentile(ticker: str, current_iv: float, days: int = 252) -> Dict[str, Any]:
    conn = get_conn()
    rows = conn.execute(
        """SELECT iv_current FROM iv_history
           WHERE ticker=? AND iv_current IS NOT NULL AND iv_current > 0
           ORDER BY recorded_date DESC LIMIT ?""",
        (ticker.upper(), days),
    ).fetchall()
    conn.close()

    if not rows:
        return {"iv_rank": None, "iv_percentile": None, "history_days": 0}

    hist = [r["iv_current"] for r in rows]
    iv_min = min(hist)
    iv_max = max(hist)

    iv_rank = (
        round((current_iv - iv_min) / (iv_max - iv_min) * 100, 1)
        if iv_max > iv_min else 50.0
    )
    iv_percentile = round(sum(1 for v in hist if v < current_iv) / len(hist) * 100, 1)

    return {
        "iv_rank": iv_rank,
        "iv_percentile": iv_percentile,
        "history_days": len(hist),
        "iv_min_52w": round(iv_min * 100, 1),
        "iv_max_52w": round(iv_max * 100, 1),
    }


def get_iv_history(ticker: str, days: int = 90) -> List[Dict]:
    conn = get_conn()
    rows = conn.execute(
        """SELECT recorded_date, iv_current, iv_30d, iv_60d, iv_90d
           FROM iv_history WHERE ticker=? AND iv_current IS NOT NULL
           ORDER BY recorded_date DESC LIMIT ?""",
        (ticker.upper(), days),
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


# ── IV Watchlist ──────────────────────────────────────────────────────────────

def get_watchlist_tickers() -> List[str]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT ticker FROM iv_watchlist WHERE is_active=1 ORDER BY added_by='builtin' DESC, added_date ASC"
    ).fetchall()
    conn.close()
    return [r["ticker"] for r in rows]


def get_watchlist_entries() -> List[Dict]:
    conn = get_conn()
    rows = conn.execute(
        "SELECT ticker, added_date, added_by, is_active FROM iv_watchlist ORDER BY added_by, ticker"
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def add_watchlist_ticker(ticker: str, added_by: str = "manual") -> bool:
    """Add ticker to watchlist. Returns True if newly inserted, False if already existed."""
    conn = get_conn()
    existing = conn.execute("SELECT ticker, is_active FROM iv_watchlist WHERE ticker=?", (ticker.upper(),)).fetchone()
    if existing:
        if not existing["is_active"]:
            conn.execute("UPDATE iv_watchlist SET is_active=1, added_by=? WHERE ticker=?", (added_by, ticker.upper()))
            conn.commit()
            conn.close()
            return True
        conn.close()
        return False
    conn.execute(
        "INSERT INTO iv_watchlist (ticker, added_date, added_by) VALUES (?, date('now'), ?)",
        (ticker.upper(), added_by)
    )
    conn.commit()
    conn.close()
    return True


def remove_watchlist_ticker(ticker: str) -> bool:
    conn = get_conn()
    conn.execute("UPDATE iv_watchlist SET is_active=0 WHERE ticker=?", (ticker.upper(),))
    changed = conn.total_changes > 0
    conn.commit()
    conn.close()
    return changed


# ── System Logs ───────────────────────────────────────────────────────────────

def log_system_event(
    level: str,
    source: str,
    message: str,
    cycle_id: Optional[str] = None,
    ticker: Optional[str] = None,
    details: Optional[Dict] = None,
) -> None:
    try:
        conn = get_conn()
        conn.execute(
            """INSERT INTO system_logs (level, source, cycle_id, ticker, message, details)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (
                level.upper(),
                source,
                cycle_id,
                ticker.upper() if ticker else None,
                message,
                json.dumps(details, default=str) if details else None,
            ),
        )
        conn.commit()
        conn.close()
    except Exception:
        pass  # Never raise from logging


def get_system_logs(
    level: Optional[str] = None,
    source: Optional[str] = None,
    cycle_id: Optional[str] = None,
    ticker: Optional[str] = None,
    date_from: Optional[str] = None,
    date_to: Optional[str] = None,
    limit: int = 300,
) -> List[Dict]:
    conn = get_conn()
    clauses = []
    params: List[Any] = []
    if level:
        clauses.append("level = ?")
        params.append(level.upper())
    if source:
        clauses.append("source LIKE ?")
        params.append(f"%{source}%")
    if cycle_id:
        clauses.append("cycle_id = ?")
        params.append(cycle_id)
    if ticker:
        clauses.append("ticker = ?")
        params.append(ticker.upper())
    if date_from:
        clauses.append("ts >= ?")
        params.append(date_from)
    if date_to:
        clauses.append("ts <= ?")
        params.append(date_to + "T23:59:59")
    where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
    rows = conn.execute(
        f"SELECT id, ts, level, source, cycle_id, ticker, message, details FROM system_logs {where} ORDER BY ts DESC LIMIT ?",
        params + [limit],
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def clear_system_logs(older_than_days: int = 30) -> int:
    conn = get_conn()
    conn.execute(f"DELETE FROM system_logs WHERE ts < datetime('now', '-{older_than_days} days')")
    deleted = conn.total_changes
    conn.commit()
    conn.close()
    return deleted


# ── Cycle Context Snapshots ───────────────────────────────────────────────────

def save_cycle_context_snapshot(run_id: str, context: dict) -> None:
    import json as _json
    conn = get_conn()
    conn.execute(
        "INSERT OR REPLACE INTO cycle_context_snapshots (run_id, context_json) VALUES (?, ?)",
        (run_id, _json.dumps(context, ensure_ascii=False, default=str)),
    )
    conn.commit()
    conn.close()


def get_cycle_context_snapshot(run_id: str) -> Optional[dict]:
    import json as _json
    conn = get_conn()
    row = conn.execute(
        "SELECT run_id, ts, context_json FROM cycle_context_snapshots WHERE run_id = ?", (run_id,)
    ).fetchone()
    conn.close()
    if not row:
        return None
    return {"run_id": row["run_id"], "ts": row["ts"], "context": _json.loads(row["context_json"])}


def list_cycle_context_snapshots(limit: int = 30) -> list:
    conn = get_conn()
    rows = conn.execute(
        "SELECT run_id, ts FROM cycle_context_snapshots ORDER BY ts DESC LIMIT ?", (limit,)
    ).fetchall()
    conn.close()
    return [{"run_id": r["run_id"], "ts": r["ts"]} for r in rows]


# ── AI Call Logs ──────────────────────────────────────────────────────────────

def save_ai_call_log(
    run_id: str,
    call_type: str,
    system_prompt: str,
    user_prompt: str,
    response_json: str,
    model: str = "gpt-4o",
    tokens_prompt: int = 0,
    tokens_completion: int = 0,
    duration_ms: int = 0,
    pattern_id: str = None,
    pattern_name: str = None,
) -> None:
    import json as _json
    conn = get_conn()
    conn.execute(
        """INSERT INTO ai_call_logs
           (run_id, call_type, pattern_id, pattern_name, system_prompt, user_prompt,
            response_json, model, tokens_prompt, tokens_completion, duration_ms)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (run_id, call_type, pattern_id, pattern_name,
         system_prompt[:8000],   # cap system prompt at 8KB
         user_prompt[:80000],    # cap user prompt at 80KB
         response_json[:50000],  # cap response at 50KB
         model, tokens_prompt, tokens_completion, duration_ms)
    )
    conn.commit()
    conn.close()


def get_ai_call_logs(run_id: str) -> list:
    import json as _json
    conn = get_conn()
    rows = conn.execute(
        """SELECT id, call_type, pattern_id, pattern_name, system_prompt, user_prompt,
                  response_json, model, tokens_prompt, tokens_completion, duration_ms, called_at
           FROM ai_call_logs WHERE run_id = ? ORDER BY called_at ASC""",
        (run_id,)
    ).fetchall()
    conn.close()
    result = []
    for r in rows:
        d = dict(r)
        # Try to parse response_json for display
        try:
            d["response"] = _json.loads(d["response_json"]) if d["response_json"] else None
        except Exception:
            d["response"] = d["response_json"]
        result.append(d)
    return result


# ── News Price Snapshots (Phase 4 — Price Discovery) ─────────────────────────

def save_news_price_snapshot(
    article_hash: str,
    article_title: str,
    ticker: str,
    expected_direction: str,
    expected_impact_score: float,
    price_at_capture: float,
    cycle_id: str,
) -> None:
    conn = get_conn()
    conn.execute(
        """INSERT OR IGNORE INTO news_price_snapshots
           (article_hash, article_title, ticker, expected_direction,
            expected_impact_score, price_at_capture, capture_cycle_id)
           VALUES (?, ?, ?, ?, ?, ?, ?)""",
        (article_hash, article_title, ticker, expected_direction,
         expected_impact_score, price_at_capture, cycle_id),
    )
    conn.commit()
    conn.close()


def get_news_price_snapshots(max_age_days: int = 7, min_age_minutes: float = 30.0) -> list:
    conn = get_conn()
    rows = conn.execute(
        """SELECT article_hash, article_title, ticker, expected_direction,
                  expected_impact_score, price_at_capture, captured_at, capture_cycle_id
           FROM news_price_snapshots
           WHERE captured_at >= datetime('now', ? || ' days')
             AND captured_at <= datetime('now', ? || ' minutes')
           ORDER BY captured_at DESC""",
        (f"-{max_age_days}", f"-{int(min_age_minutes)}"),
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def purge_old_price_snapshots(older_than_days: int = 14) -> int:
    conn = get_conn()
    conn.execute(
        "DELETE FROM news_price_snapshots WHERE captured_at < datetime('now', ? || ' days')",
        (f"-{older_than_days}",),
    )
    deleted = conn.total_changes
    conn.commit()
    conn.close()
    return deleted


# ── Knowledge Base Decay ──────────────────────────────────────────────────────

def decay_kb_confidence() -> int:
    """
    Decrease confidence on KB entries past their expires_at or older than
    confidence_decay_days since last_confirmed_at. Archives entries at 0.
    Returns number of entries updated.
    """
    conn = get_conn()
    c = conn.cursor()
    today_str = datetime.utcnow().date().isoformat()

    # Entries past expires_at → archive
    c.execute("""
        UPDATE knowledge_base
        SET status = 'archived', confidence = 0
        WHERE expires_at IS NOT NULL AND expires_at <= ? AND status = 'active'
    """, (today_str,))
    expired = c.rowcount

    # Entries where days_since_confirmation > confidence_decay_days
    # Reduce confidence by 10 per overdue period
    rows = c.execute("""
        SELECT id, confidence, last_confirmed_at, confidence_decay_days
        FROM knowledge_base
        WHERE status = 'active' AND last_confirmed_at IS NOT NULL
    """).fetchall()

    decayed = 0
    for row in rows:
        r = dict(row)
        try:
            from datetime import date as _d
            last = _d.fromisoformat(r["last_confirmed_at"][:10])
            days_since = (_d.today() - last).days
            decay_period = r["confidence_decay_days"] or 90
            if days_since > decay_period:
                periods_overdue = days_since // decay_period
                new_conf = max(0, r["confidence"] - periods_overdue * 10)
                if new_conf != r["confidence"]:
                    c.execute(
                        "UPDATE knowledge_base SET confidence=? WHERE id=?",
                        (new_conf, r["id"])
                    )
                    decayed += 1
                    if new_conf == 0:
                        c.execute(
                            "UPDATE knowledge_base SET status='archived' WHERE id=?",
                            (r["id"],)
                        )
        except Exception:
            pass

    conn.commit()
    conn.close()
    return expired + decayed


# ── Pattern Reliability ───────────────────────────────────────────────────────

def get_pattern_reliability(pattern_id: str = None) -> List[Dict]:
    """
    Compute win_rate, avg_pnl, trade_count, reliability_score per pattern.
    Uses MATURE trades only (days_held >= 35% of horizon_days).
    If pattern_id is provided, returns single-item list for that pattern.
    """
    import math
    from datetime import date as _date

    conn = get_conn()
    q = "SELECT * FROM trade_entry_prices WHERE pnl_pct IS NOT NULL"
    args: list = []
    if pattern_id:
        q += " AND pattern_id = ?"
        args.append(pattern_id)
    rows = conn.execute(q, args).fetchall()
    conn.close()

    today = _date.today()
    mature_threshold = float(get_config("maturity_threshold_pct") or "35") / 100.0
    by_pattern: Dict[str, list] = {}
    for row in rows:
        r = dict(row)
        try:
            entry = _date.fromisoformat(r["entry_date"])
            days_held = (today - entry).days
        except Exception:
            days_held = 0
        horizon = r.get("horizon_days") or 30
        ratio = days_held / horizon if horizon else 0
        if ratio < mature_threshold:
            continue
        by_pattern.setdefault(r["pattern_id"], []).append(r)

    result = []
    for pid, trades in by_pattern.items():
        pnls = [t["pnl_pct"] for t in trades if t.get("pnl_pct") is not None]
        if not pnls:
            continue
        wins = sum(1 for p in pnls if p > 0)
        win_rate = wins / len(pnls)
        avg_pnl = sum(pnls) / len(pnls)
        # Composite score: win_rate × log(n+1) — penalises small samples
        reliability = round(win_rate * math.log(len(pnls) + 1), 3)

        result.append({
            "pattern_id": pid,
            "pattern_name": trades[0].get("pattern_name", pid),
            "trade_count": len(pnls),
            "win_rate": round(win_rate, 3),
            "win_rate_pct": round(win_rate * 100, 1),
            "avg_pnl_pct": round(avg_pnl, 2),
            "max_pnl_pct": round(max(pnls), 2),
            "max_loss_pct": round(min(pnls), 2),
            "reliability_score": reliability,
        })

    result.sort(key=lambda x: -x["reliability_score"])
    return result


def get_all_pattern_reliability_map() -> Dict[str, Dict]:
    """Returns {pattern_id: reliability_dict} for fast lookup."""
    return {r["pattern_id"]: r for r in get_pattern_reliability()}


# ── Calibration ───────────────────────────────────────────────────────────────

def get_calibration_data(days: int = 365) -> Dict:
    """
    Compare predicted probability (stored at score time) vs realized outcome
    (pnl_pct > 0 at maturity) to compute Brier score and calibration buckets.
    """
    import math
    from datetime import date as _date

    conn = get_conn()
    # Join pattern_scores (has predicted probability) with trade_entry_prices (has realized P&L)
    rows = conn.execute("""
        SELECT
            psh.pattern_id,
            psh.score,
            tep.pnl_pct,
            tep.entry_date,
            tep.horizon_days,
            cp.probability as predicted_prob
        FROM pattern_score_history psh
        JOIN trade_entry_prices tep ON tep.pattern_id = psh.pattern_id
        LEFT JOIN custom_patterns cp ON cp.id = psh.pattern_id
        WHERE tep.pnl_pct IS NOT NULL
          AND cp.probability IS NOT NULL
          AND tep.entry_date >= date('now', ?)
    """, (f"-{days} days",)).fetchall()
    conn.close()

    today = _date.today()
    pairs = []
    for row in rows:
        r = dict(row)
        try:
            entry = _date.fromisoformat(r["entry_date"])
            dh = (today - entry).days
        except Exception:
            dh = 0
        horizon = r.get("horizon_days") or 30
        if dh / horizon < 0.35:
            continue  # only mature
        pred = float(r["predicted_prob"] or 0)
        realized = 1.0 if (r["pnl_pct"] or 0) > 0 else 0.0
        pairs.append({"predicted": pred, "realized": realized})

    if not pairs:
        return {"pairs": [], "brier_score": None, "buckets": [], "sample_size": 0}

    # Brier score
    brier = sum((p["predicted"] - p["realized"]) ** 2 for p in pairs) / len(pairs)

    # Calibration buckets (deciles)
    buckets = []
    for low in [i / 10 for i in range(0, 10)]:
        high = low + 0.1
        bucket_pairs = [p for p in pairs if low <= p["predicted"] < high]
        if bucket_pairs:
            actual_rate = sum(p["realized"] for p in bucket_pairs) / len(bucket_pairs)
            buckets.append({
                "predicted_range": f"{int(low*100)}-{int(high*100)}%",
                "predicted_mid": round((low + high) / 2, 2),
                "actual_rate": round(actual_rate, 3),
                "count": len(bucket_pairs),
                "bias": round(actual_rate - (low + high) / 2, 3),
            })

    return {
        "pairs": pairs,
        "brier_score": round(brier, 4),
        "buckets": buckets,
        "sample_size": len(pairs),
        "interpretation": (
            "Bien calibré" if brier < 0.15
            else "Modérément calibré" if brier < 0.25
            else "Surconfiant ou mal calibré"
        ),
    }


# ╔══════════════════════════════════════════════════════════════════════════════╗
# ║  PHASE 3 — Portfolio Risk Engine                                            ║
# ╚══════════════════════════════════════════════════════════════════════════════╝

# Risk factor → (asset_classes, trigger_keywords)
_RISK_FACTOR_MAP = {
    "géopolitique": {
        "asset_classes": {"energy", "metals", "agriculture"},
        "triggers": {"military", "sanctions", "trade_war", "political_speech"},
    },
    "inflation": {
        "asset_classes": {"energy", "metals", "agriculture", "forex"},
        "triggers": {"energy", "resource_scarcity", "trade_war"},
    },
    "récession": {
        "asset_classes": {"indices", "equities", "rates"},
        "triggers": {"financial_crisis", "elections"},
    },
    "liquidité": {
        "asset_classes": {"indices", "equities", "rates"},
        "triggers": {"financial_crisis", "health_crisis"},
    },
    "dollar": {
        "asset_classes": {"forex", "metals"},
        "triggers": {"sanctions", "financial_crisis", "trade_war"},
    },
}


def _classify_risk_factors(asset_class: str, triggers: List[str]) -> List[str]:
    """Return list of risk factors for a trade given its asset_class and pattern triggers."""
    ac = (asset_class or "").lower()
    trg_set = {t.lower() for t in (triggers or [])}
    factors = []
    for factor, cfg in _RISK_FACTOR_MAP.items():
        if ac in cfg["asset_classes"] or trg_set & cfg["triggers"]:
            factors.append(factor)
    return factors or ["autre"]


# ── Sprint 3.1 — Portfolio Exposure ──────────────────────────────────────────

def get_portfolio_exposure() -> Dict:
    """
    Returns exposure by asset class and by risk factor for open positions,
    plus P&L timeline and concentration alerts.
    """
    conn = get_conn()
    trades = conn.execute(
        "SELECT * FROM portfolio WHERE status='open'"
    ).fetchall()
    conn.close()

    by_class: Dict[str, Dict] = {}
    by_factor: Dict[str, Dict] = {}
    total_capital = 0.0

    for row in trades:
        t = dict(row)
        ac = (t.get("asset_class") or "autre").lower()
        cap = float(t.get("capital_invested") or 0)
        total_capital += cap

        if ac not in by_class:
            by_class[ac] = {"capital": 0.0, "trade_count": 0, "trades": []}
        by_class[ac]["capital"] += cap
        by_class[ac]["trade_count"] += 1
        by_class[ac]["trades"].append(t.get("id"))

        # Pattern triggers for risk factor classification
        triggers: List[str] = []
        try:
            conn2 = get_conn()
            pat_row = conn2.execute(
                "SELECT triggers FROM custom_patterns WHERE id=?",
                (t.get("geo_trigger") or "",)
            ).fetchone()
            conn2.close()
            if pat_row and pat_row["triggers"]:
                triggers = json.loads(pat_row["triggers"] or "[]")
        except Exception:
            pass

        factors = _classify_risk_factors(ac, triggers)
        for f in factors:
            if f not in by_factor:
                by_factor[f] = {"capital": 0.0, "trade_count": 0, "trades": []}
            by_factor[f]["capital"] += cap
            by_factor[f]["trade_count"] += 1
            by_factor[f]["trades"].append(t.get("id"))

    # Compute percentages + concentration alerts
    alerts = []
    for ac, info in by_class.items():
        pct = round(info["capital"] / total_capital * 100, 1) if total_capital else 0
        info["pct_of_portfolio"] = pct
        if pct > 40:
            alerts.append({
                "type": "concentration_class",
                "level": "high" if pct > 60 else "warning",
                "message": f"Concentration élevée sur {ac.upper()}: {pct}% du capital",
                "asset_class": ac,
                "pct": pct,
            })

    for factor, info in by_factor.items():
        pct = round(info["capital"] / total_capital * 100, 1) if total_capital else 0
        info["pct_of_portfolio"] = pct
        if pct > 50:
            alerts.append({
                "type": "concentration_factor",
                "level": "high" if pct > 70 else "warning",
                "message": f"Risque concentré sur facteur '{factor}': {pct}% du capital",
                "factor": factor,
                "pct": pct,
            })

    return {
        "by_class": by_class,
        "by_factor": by_factor,
        "total_capital": total_capital,
        "open_trade_count": len(trades),
        "concentration_alerts": alerts,
    }


def get_pnl_timeline(days: int = 90) -> List[Dict]:
    """
    Returns daily aggregated P&L from trade_entry_prices (closed/mature trades).
    Used for portfolio equity curve.
    """
    conn = get_conn()
    rows = conn.execute("""
        SELECT entry_date, SUM(pnl_pct * capital_invested / 100) as daily_pnl_abs,
               AVG(pnl_pct) as avg_pnl_pct, COUNT(*) as trade_count
        FROM trade_entry_prices
        WHERE pnl_pct IS NOT NULL AND entry_date >= date('now', ?)
        GROUP BY entry_date
        ORDER BY entry_date ASC
    """, (f"-{days} days",)).fetchall()
    conn.close()

    cumulative = 0.0
    result = []
    for row in rows:
        r = dict(row)
        cumulative += r.get("daily_pnl_abs") or 0
        r["cumulative_pnl_abs"] = round(cumulative, 2)
        result.append(r)
    return result


# ── Sprint 3.2 — Risk Cluster Engine ────────────────────────────────────────

def get_risk_clusters() -> Dict:
    """
    Classify all open trades by risk factor, compute exposure per factor,
    detect saturation (>50%), and return cluster data for the scoring prompt.
    """
    exposure = get_portfolio_exposure()
    by_factor = exposure["by_factor"]
    total = exposure["total_capital"]

    clusters = []
    saturated_factors = []
    for factor, info in by_factor.items():
        pct = info.get("pct_of_portfolio", 0)
        saturated = pct > 50
        if saturated:
            saturated_factors.append(factor)
        clusters.append({
            "factor": factor,
            "capital": round(info["capital"], 2),
            "pct_of_portfolio": pct,
            "trade_count": info["trade_count"],
            "saturated": saturated,
        })

    clusters.sort(key=lambda x: -x["pct_of_portfolio"])

    return {
        "clusters": clusters,
        "saturated_factors": saturated_factors,
        "total_capital": total,
        "risk_prompt_context": _build_risk_cluster_prompt(clusters, saturated_factors),
    }


def _build_risk_cluster_prompt(clusters: List[Dict], saturated: List[str]) -> str:
    if not clusters:
        return ""
    lines = ["## ⚠ ÉTAT DU PORTEFEUILLE — Concentration des risques"]
    for c in clusters[:5]:
        sat_mark = " 🔴 SATURÉ" if c["saturated"] else ""
        lines.append(f"  - Facteur '{c['factor']}': {c['pct_of_portfolio']}% du capital ({c['trade_count']} trades){sat_mark}")
    if saturated:
        lines.append(
            f"\n⚠ CONSIGNE SCORING: Les facteurs [{', '.join(saturated)}] sont SATURÉS. "
            "Pénaliser de 15 points les patterns dépendants de ces facteurs. "
            "Favoriser les patterns sur d'autres facteurs pour diversifier."
        )
    return "\n".join(lines)


def get_pattern_correlations() -> Dict:
    """
    Compute Pearson correlation of P&L between pattern pairs with ≥3 mature trades each.
    Returns correlation matrix and sorted pair list.
    """
    import math
    from datetime import date as _d

    conn = get_conn()
    rows = conn.execute("""
        SELECT tep.pattern_id, tep.pnl_pct, tep.entry_date, tep.horizon_days,
               cp.name as pattern_name
        FROM trade_entry_prices tep
        LEFT JOIN custom_patterns cp ON cp.id = tep.pattern_id
        WHERE tep.pnl_pct IS NOT NULL
    """).fetchall()
    conn.close()

    today = _d.today()
    by_pattern: Dict[str, list] = {}
    names: Dict[str, str] = {}

    for row in rows:
        r = dict(row)
        pid = r["pattern_id"]
        try:
            entry = _d.fromisoformat(r["entry_date"])
            days_held = (today - entry).days
        except Exception:
            days_held = 0
        horizon = r.get("horizon_days") or 30
        if days_held / horizon < 0.35:
            continue  # mature only
        by_pattern.setdefault(pid, []).append(float(r["pnl_pct"]))
        names[pid] = r.get("pattern_name") or pid

    # Keep patterns with ≥3 trades
    patterns = {pid: pnls for pid, pnls in by_pattern.items() if len(pnls) >= 3}

    def pearson(a: list, b: list) -> Optional[float]:
        n = min(len(a), len(b))
        if n < 2:
            return None
        xa, xb = a[:n], b[:n]
        ma, mb = sum(xa) / n, sum(xb) / n
        num = sum((xa[i] - ma) * (xb[i] - mb) for i in range(n))
        da = math.sqrt(sum((x - ma) ** 2 for x in xa))
        db = math.sqrt(sum((x - mb) ** 2 for x in xb))
        if da * db == 0:
            return None
        return round(num / (da * db), 3)

    pids = list(patterns.keys())
    pairs = []
    matrix: Dict[str, Dict[str, Optional[float]]] = {}

    for i, pa in enumerate(pids):
        matrix[pa] = {}
        for pb in pids:
            if pa == pb:
                matrix[pa][pb] = 1.0
            else:
                corr = pearson(patterns[pa], patterns[pb])
                matrix[pa][pb] = corr
        for pb in pids[i + 1:]:
            corr = matrix[pa].get(pb)
            if corr is not None:
                pairs.append({
                    "pattern_a": pa,
                    "name_a": names.get(pa, pa),
                    "pattern_b": pb,
                    "name_b": names.get(pb, pb),
                    "correlation": corr,
                    "interpretation": (
                        "Très corrélés — risque concentré" if abs(corr) > 0.7
                        else "Modérément corrélés" if abs(corr) > 0.4
                        else "Faiblement corrélés"
                    ),
                })

    pairs.sort(key=lambda x: -abs(x["correlation"]))

    return {
        "matrix": matrix,
        "pairs": pairs[:20],
        "pattern_names": names,
        "pattern_count": len(pids),
    }


# ── Sprint 3.3 — Kelly Fractional Sizing ────────────────────────────────────

def compute_kelly_sizing(
    pattern_id: str,
    capital_available: float = 10000.0,
    fractional: float = 0.33,
) -> Dict:
    """
    Compute fractional Kelly position sizing for a pattern.
    f* = (p × G - (1-p)) / G  where G = expected gain as multiplier.
    Fractional Kelly = fractional × f* (default 33% = between 25-50% institutional norm).
    Adjusted for risk cluster saturation.
    """
    conn = get_conn()
    pat = conn.execute("SELECT * FROM custom_patterns WHERE id=?", (pattern_id,)).fetchone()
    conn.close()
    if not pat:
        return {"error": "Pattern not found"}

    p = dict(pat)
    prob = float(p.get("probability") or 0.5)
    expected_move = float(p.get("expected_move_pct") or 50) / 100  # as decimal

    # G = gain multiplier (if trade wins, return expected_move; if loses, -1)
    G = max(expected_move, 0.01)
    kelly_full = (prob * G - (1 - prob)) / G
    kelly_full = max(0.0, kelly_full)  # never negative

    kelly_frac = kelly_full * fractional

    # Risk cluster adjustment: halve if saturated
    ac = (p.get("asset_class") or "").lower()
    triggers_raw = p.get("triggers") or "[]"
    try:
        triggers_list = json.loads(triggers_raw) if isinstance(triggers_raw, str) else triggers_raw
    except Exception:
        triggers_list = []
    factors = _classify_risk_factors(ac, triggers_list)

    clusters = get_risk_clusters()
    saturated = set(clusters.get("saturated_factors", []))
    cluster_adjusted = kelly_frac
    cluster_adjustment_reason = None

    if factors and saturated & set(factors):
        cluster_adjusted = kelly_frac / 2
        cluster_adjustment_reason = f"Facteur {'|'.join(saturated & set(factors))} saturé → sizing ÷ 2"

    suggested_capital = round(capital_available * cluster_adjusted, 2)
    suggested_capital_display = min(suggested_capital, capital_available * 0.25)  # hard cap 25%

    # Reliability adjustment (if available)
    reliability = get_pattern_reliability(pattern_id=pattern_id)
    reliability_adjustment = None
    if reliability:
        rel = reliability[0]
        if rel["trade_count"] >= 5 and rel["win_rate"] < 0.4:
            suggested_capital_display *= 0.5
            reliability_adjustment = f"Win rate historique faible ({rel['win_rate_pct']}%) → sizing ÷ 2"

    return {
        "pattern_id": pattern_id,
        "pattern_name": p.get("name"),
        "probability": prob,
        "expected_move_pct": round(expected_move * 100, 1),
        "kelly_full": round(kelly_full, 4),
        "kelly_fractional": round(kelly_frac, 4),
        "fractional_pct": round(fractional * 100),
        "cluster_adjusted_kelly": round(cluster_adjusted, 4),
        "risk_factors": factors,
        "saturated_factors": list(saturated & set(factors)),
        "cluster_adjustment_reason": cluster_adjustment_reason,
        "reliability_adjustment": reliability_adjustment,
        "suggested_capital_pct": round(cluster_adjusted * 100, 1),
        "suggested_capital_eur": round(suggested_capital_display, 0),
        "capital_available": capital_available,
        "explanation": (
            f"Kelly complet = {kelly_full*100:.1f}% → Kelly fractionnel ({fractional*100:.0f}%) "
            f"= {kelly_frac*100:.1f}%"
            + (f" → Ajusté cluster = {cluster_adjusted*100:.1f}%" if cluster_adjustment_reason else "")
        ),
    }


# ── Sprint 3.4 — Risk Dashboard ──────────────────────────────────────────────

def get_risk_dashboard() -> Dict:
    """
    Full portfolio risk snapshot: concentration, diversification score,
    expected drawdown estimate, and auto-recommendation.
    """
    import math

    exposure = get_portfolio_exposure()
    clusters = get_risk_clusters()
    corr_data = get_pattern_correlations()

    by_class = exposure["by_class"]
    by_factor = exposure["by_factor"]
    total = exposure["total_capital"]
    alerts = exposure["concentration_alerts"]

    # Herfindahl-Hirschman Index (HHI) as concentration measure
    # HHI = sum of (share_i)^2. HHI=1 fully concentrated, HHI=1/N fully diversified
    shares = [info["pct_of_portfolio"] / 100 for info in by_class.values() if info.get("pct_of_portfolio")]
    hhi = sum(s ** 2 for s in shares) if shares else 0
    n_classes = len(shares)
    hhi_min = 1 / n_classes if n_classes > 0 else 1

    # Effective N = 1/HHI (Herfindahl diversity = N effective independent positions)
    effective_n = round(1 / hhi, 2) if hhi > 0 else n_classes
    max_possible_n = n_classes if n_classes > 0 else 1
    diversification_score = round(min(effective_n / max(max_possible_n, 1), 1.0) * 100, 1)

    # Expected drawdown estimate: weighted by concentration
    # Simple model: if one factor is at X% and has 40% historical loss → max drawdown = X% × 40%
    FACTOR_DRAWDOWN = {
        "géopolitique": 0.40,  # high volatility
        "inflation": 0.30,
        "récession": 0.45,
        "liquidité": 0.35,
        "dollar": 0.25,
        "autre": 0.30,
    }
    expected_drawdown_pct = 0.0
    for factor, info in by_factor.items():
        w = info.get("pct_of_portfolio", 0) / 100
        dd = FACTOR_DRAWDOWN.get(factor, 0.30)
        expected_drawdown_pct += w * dd * 100

    # High-correlation pairs count
    high_corr_pairs = [p for p in corr_data.get("pairs", []) if abs(p["correlation"]) > 0.7]

    # Auto-recommendation
    recommendation = _build_risk_recommendation(
        clusters.get("saturated_factors", []),
        diversification_score,
        round(expected_drawdown_pct, 1),
        alerts,
        high_corr_pairs,
    )

    return {
        "exposure_by_class": {k: {**v, "pct_of_portfolio": v.get("pct_of_portfolio", 0)} for k, v in by_class.items()},
        "exposure_by_factor": {k: {**v, "pct_of_portfolio": v.get("pct_of_portfolio", 0)} for k, v in by_factor.items()},
        "total_capital": total,
        "open_trades": exposure["open_trade_count"],
        "hhi": round(hhi, 4),
        "diversification_score": diversification_score,
        "effective_n_positions": effective_n,
        "expected_drawdown_pct": round(expected_drawdown_pct, 1),
        "concentration_alerts": alerts,
        "high_correlation_pairs": high_corr_pairs[:5],
        "saturated_factors": clusters.get("saturated_factors", []),
        "risk_clusters": clusters.get("clusters", []),
        "recommendation": recommendation,
    }


# ╔══════════════════════════════════════════════════════════════════════════════╗
# ║  PHASE 4 — Moteur Probabiliste & Apprentissage Automatique                  ║
# ╚══════════════════════════════════════════════════════════════════════════════╝

# ── Sprint 4.1 — Bayesian Updating ───────────────────────────────────────────

def update_bayesian_posteriors() -> int:
    """
    Met à jour les posteriors Beta(α,β) de chaque pattern selon ses trades matures.
    Prior faible : α₀=1, β₀=1 (Laplace smoothing).
    Posterior : α = 1 + wins, β = 1 + losses → win_rate bayésien = α/(α+β)
    Retourne le nombre de patterns mis à jour.
    """
    import math as _math
    from datetime import date as _date

    conn = get_conn()
    rows = conn.execute(
        "SELECT pattern_id, pnl_pct, entry_date, horizon_days FROM trade_entry_prices WHERE pnl_pct IS NOT NULL"
    ).fetchall()
    conn.close()

    today = _date.today()
    by_pattern: Dict[str, list] = {}
    for row in rows:
        r = dict(row)
        try:
            entry = _date.fromisoformat(r["entry_date"])
            days_held = (today - entry).days
        except Exception:
            continue
        horizon = r.get("horizon_days") or 30
        if days_held / max(horizon, 1) < 0.35:
            continue  # trades immatures exclus
        by_pattern.setdefault(r["pattern_id"], []).append(float(r["pnl_pct"] or 0))

    if not by_pattern:
        return 0

    conn = get_conn()
    c = conn.cursor()
    updated = 0
    now_iso = datetime.utcnow().isoformat()
    for pid, pnls in by_pattern.items():
        n = len(pnls)
        wins = sum(1 for p in pnls if p > 0)
        losses = n - wins
        alpha = 1.0 + wins   # posterior alpha
        beta  = 1.0 + losses # posterior beta
        bayes_wr = alpha / (alpha + beta)
        c.execute("""
            UPDATE custom_patterns
            SET bayesian_alpha=?, bayesian_beta=?, bayesian_win_rate=?,
                bayesian_updated_at=?, bayesian_sample_size=?
            WHERE id=?
        """, (round(alpha, 1), round(beta, 1), round(bayes_wr, 4), now_iso, n, pid))
        if c.rowcount:
            updated += 1
    conn.commit()
    conn.close()
    return updated


def get_bayesian_posteriors() -> List[Dict]:
    """
    Retourne tous les patterns avec leurs posteriors bayésiens + intervalle de crédibilité 95%.
    CI 95% via approximation normale : ±1.96 × σ = ±1.96 × sqrt(αβ / (α+β)²(α+β+1))
    """
    import math as _math
    conn = get_conn()
    rows = conn.execute("""
        SELECT id, name, probability, bayesian_alpha, bayesian_beta,
               bayesian_win_rate, bayesian_sample_size, bayesian_updated_at,
               asset_class, is_active
        FROM custom_patterns
        WHERE is_active=1
        ORDER BY bayesian_sample_size DESC, name
    """).fetchall()
    conn.close()

    result = []
    for row in rows:
        r = dict(row)
        alpha = r.get("bayesian_alpha") or 1.0
        beta  = r.get("bayesian_beta")  or 1.0
        n = alpha + beta
        # 95% credible interval (Beta distribution approximation)
        variance = (alpha * beta) / (n * n * (n + 1))
        std = _math.sqrt(max(variance, 0))
        bayes_wr = r.get("bayesian_win_rate") or alpha / n
        lower = max(0.0, bayes_wr - 1.96 * std)
        upper = min(1.0, bayes_wr + 1.96 * std)
        sample_size = r.get("bayesian_sample_size") or 0

        # Écart entre prior GPT et posterior bayésien
        prior = r.get("probability") or 0.5
        drift = round(bayes_wr - prior, 3)

        result.append({
            "pattern_id": r["id"],
            "pattern_name": r["name"],
            "asset_class": r.get("asset_class"),
            "prior_probability": round(prior, 3),
            "bayesian_win_rate": round(bayes_wr, 3),
            "bayesian_win_rate_pct": round(bayes_wr * 100, 1),
            "lower_ci_95": round(lower, 3),
            "upper_ci_95": round(upper, 3),
            "ci_width": round(upper - lower, 3),
            "sample_size": sample_size,
            "alpha": alpha,
            "beta": beta,
            "prior_vs_posterior_drift": drift,
            "updated_at": r.get("bayesian_updated_at"),
            "confidence_level": (
                "haute" if sample_size >= 10
                else "moyenne" if sample_size >= 5
                else "faible"
            ),
        })
    return result


# ── Sprint 4.2 — Détection automatique de régimes ────────────────────────────

_GAUGE_FEATURES = [
    "vix", "slope_10y3m", "dxy", "brent", "gold", "copper", "spx_vs_200d"
]

_CLUSTER_LABELS = {
    0: "Stress Géopolitique",
    1: "Expansion Tranquille",
    2: "Récession / Risk-Off",
    3: "Stagflation / Inflation",
    4: "Transition / Incertain",
}


def _kmeans_numpy(X, n_clusters: int = 4, max_iter: int = 100, seed: int = 42):
    """K-Means minimal en numpy pur (pas de sklearn nécessaire)."""
    import numpy as np
    rng = np.random.default_rng(seed)
    idx = rng.choice(len(X), size=n_clusters, replace=False)
    centroids = X[idx].copy()

    for _ in range(max_iter):
        dists = np.linalg.norm(X[:, None, :] - centroids[None, :, :], axis=2)
        labels = np.argmin(dists, axis=1)
        new_centroids = np.array([
            X[labels == k].mean(axis=0) if (labels == k).any() else centroids[k]
            for k in range(n_clusters)
        ])
        if np.allclose(centroids, new_centroids, atol=1e-6):
            break
        centroids = new_centroids
    return labels, centroids


def detect_and_save_regime_clusters(n_clusters: int = 4, days: int = 180) -> Dict:
    """
    Applique K-Means sur l'historique des gauges macro (VIX, pente, DXY…).
    Sauvegarde l'assignation courante dans regime_clusters.
    Retourne le cluster actuel + les centroïdes labellisés.
    """
    import numpy as np
    import json as _json

    conn = get_conn()
    rows = conn.execute("""
        SELECT timestamp, dominant, gauges_summary_json
        FROM macro_regime_history
        WHERE timestamp >= datetime('now', ?)
        ORDER BY timestamp ASC
    """, (f"-{days} days",)).fetchall()
    conn.close()

    if len(rows) < max(n_clusters * 2, 8):
        return {"error": "Données insuffisantes", "min_required": max(n_clusters * 2, 8)}

    snapshots = []
    timestamps = []
    dominants = []
    for row in rows:
        r = dict(row)
        try:
            gauges = _json.loads(r.get("gauges_summary_json") or "{}")
        except Exception:
            continue
        vec = []
        for feat in _GAUGE_FEATURES:
            g = gauges.get(feat, {})
            val = g.get("value") if isinstance(g, dict) else g
            vec.append(float(val) if val is not None else 0.0)
        snapshots.append(vec)
        timestamps.append(r["timestamp"])
        dominants.append(r.get("dominant", "incertain"))

    if not snapshots:
        return {"error": "Aucune donnée de gauges exploitable"}

    X = np.array(snapshots, dtype=float)
    # Normalisation z-score par feature
    mean = X.mean(axis=0)
    std  = X.std(axis=0)
    std[std == 0] = 1.0
    X_norm = (X - mean) / std

    # Anomaly detection : points > 3σ du vecteur global
    global_dist = np.linalg.norm(X_norm, axis=1)
    anomaly_threshold = global_dist.mean() + 3 * global_dist.std()

    labels, centroids = _kmeans_numpy(X_norm, n_clusters=n_clusters)

    # Labelliser les clusters selon le dominant le plus fréquent
    cluster_regimes: Dict[int, str] = {}
    for k in range(n_clusters):
        idxs = [i for i, l in enumerate(labels) if l == k]
        if idxs:
            dom_counts: Dict[str, int] = {}
            for i in idxs:
                d = dominants[i]
                dom_counts[d] = dom_counts.get(d, 0) + 1
            cluster_regimes[k] = max(dom_counts, key=dom_counts.get)

    # Sauvegarder le snapshot le plus récent
    last_idx = len(labels) - 1
    current_cluster = int(labels[last_idx])
    current_anomaly = bool(global_dist[last_idx] > anomaly_threshold)

    conn = get_conn()
    conn.execute("""
        INSERT INTO regime_clusters
            (timestamp, cluster_id, cluster_label, dominant_regime, gauges_json, anomaly_flag)
        VALUES (?, ?, ?, ?, ?, ?)
    """, (
        timestamps[last_idx],
        current_cluster,
        _CLUSTER_LABELS.get(current_cluster, f"Cluster {current_cluster}"),
        cluster_regimes.get(current_cluster, dominants[last_idx]),
        _json.dumps({f: round(float(snapshots[last_idx][i]), 3) for i, f in enumerate(_GAUGE_FEATURES)}),
        int(current_anomaly),
    ))
    conn.commit()
    conn.close()

    # Statistiques par cluster
    cluster_stats = {}
    for k in range(n_clusters):
        k_idxs = [i for i, l in enumerate(labels) if l == k]
        cluster_stats[k] = {
            "cluster_id": k,
            "label": _CLUSTER_LABELS.get(k, f"Cluster {k}"),
            "dominant_regime": cluster_regimes.get(k, "incertain"),
            "count": len(k_idxs),
            "pct": round(len(k_idxs) / len(labels) * 100, 1),
            "centroid": {f: round(float(centroids[k][i]), 3) for i, f in enumerate(_GAUGE_FEATURES)},
        }

    return {
        "current_cluster": current_cluster,
        "current_label": _CLUSTER_LABELS.get(current_cluster, f"Cluster {current_cluster}"),
        "current_anomaly": current_anomaly,
        "cluster_stats": list(cluster_stats.values()),
        "n_snapshots": len(labels),
        "features": _GAUGE_FEATURES,
    }


def get_regime_cluster_history(days: int = 90) -> List[Dict]:
    """Retourne l'historique des assignations de clusters."""
    conn = get_conn()
    rows = conn.execute("""
        SELECT timestamp, cluster_id, cluster_label, dominant_regime, anomaly_flag, created_at
        FROM regime_clusters
        WHERE timestamp >= datetime('now', ?)
        ORDER BY timestamp DESC
        LIMIT 200
    """, (f"-{days} days",)).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_regime_transition_matrix(days: int = 180) -> Dict:
    """
    Calcule la matrice de transition entre clusters :
    P(cluster_j | cluster_i) = nb transitions i→j / nb fois où on était en i.
    """
    rows = get_regime_cluster_history(days=days)
    if len(rows) < 4:
        return {"matrix": {}, "labels": {}, "n_transitions": 0}

    # rows trié DESC → inverser pour avoir l'ordre chronologique
    seq = list(reversed(rows))
    cluster_ids = sorted({r["cluster_id"] for r in seq})

    counts: Dict[int, Dict[int, int]] = {c: {d: 0 for d in cluster_ids} for c in cluster_ids}
    totals: Dict[int, int] = {c: 0 for c in cluster_ids}

    for i in range(len(seq) - 1):
        src = seq[i]["cluster_id"]
        dst = seq[i + 1]["cluster_id"]
        counts[src][dst] = counts[src].get(dst, 0) + 1
        totals[src] = totals.get(src, 0) + 1

    matrix = {}
    for src in cluster_ids:
        matrix[src] = {}
        total = totals.get(src, 0)
        for dst in cluster_ids:
            matrix[src][dst] = round(counts[src].get(dst, 0) / total, 3) if total else 0.0

    labels = {k: _CLUSTER_LABELS.get(k, f"Cluster {k}") for k in cluster_ids}
    n_trans = sum(totals.values())

    return {"matrix": matrix, "labels": labels, "cluster_ids": cluster_ids, "n_transitions": n_trans}


# ── Sprint 4.3 — Embeddings sémantiques ──────────────────────────────────────

def _cosine_similarity(a: List[float], b: List[float]) -> float:
    """Similarité cosinus entre deux vecteurs."""
    import math as _math
    dot = sum(x * y for x, y in zip(a, b))
    na = _math.sqrt(sum(x * x for x in a))
    nb = _math.sqrt(sum(y * y for y in b))
    if na == 0 or nb == 0:
        return 0.0
    return dot / (na * nb)


def get_or_create_pattern_embedding(pattern_id: str, text: str, api_key: str) -> Optional[List[float]]:
    """
    Retourne l'embedding stocké pour ce pattern, ou le crée via OpenAI text-embedding-3-small.
    Le vecteur est stocké en JSON dans pattern_embeddings.
    """
    import json as _json
    conn = get_conn()
    row = conn.execute(
        "SELECT embedding_json FROM pattern_embeddings WHERE pattern_id=?", (pattern_id,)
    ).fetchone()
    conn.close()

    if row:
        try:
            return _json.loads(row["embedding_json"])
        except Exception:
            pass

    # Créer via OpenAI
    try:
        import urllib.request as _req
        import urllib.error as _uerr
        payload = _json.dumps({
            "input": text[:8000],
            "model": "text-embedding-3-small",
        }).encode("utf-8")
        request = _req.Request(
            "https://api.openai.com/v1/embeddings",
            data=payload,
            headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"},
            method="POST",
        )
        with _req.urlopen(request, timeout=15) as resp:
            data = _json.loads(resp.read())
        vec = data["data"][0]["embedding"]
    except Exception:
        return None

    # Persister
    vec_json = _json.dumps(vec)
    conn = get_conn()
    conn.execute("""
        INSERT OR REPLACE INTO pattern_embeddings (pattern_id, embedding_json, model_version, updated_at)
        VALUES (?, ?, 'text-embedding-3-small', ?)
    """, (pattern_id, vec_json, datetime.utcnow().isoformat()))
    conn.commit()
    conn.close()
    return vec


def max_cosine_similarity_vs_existing(
    candidate_text: str,
    existing_patterns: List[Dict],
    api_key: str,
    candidate_id: Optional[str] = None,
) -> float:
    """
    Calcule la similarité cosinus maximale entre le candidat et les patterns existants.
    Retourne 0.0 si les embeddings ne sont pas disponibles (fallback Jaccard dans l'appelant).
    """
    import json as _json
    if not api_key or not existing_patterns:
        return 0.0

    # Obtenir l'embedding du candidat
    _tmp_id = candidate_id or f"_tmp_{hash(candidate_text) & 0xFFFFFF}"
    cand_vec = get_or_create_pattern_embedding(_tmp_id, candidate_text, api_key)
    if not cand_vec:
        return 0.0

    max_sim = 0.0
    conn = get_conn()
    for p in existing_patterns:
        pid = p.get("id")
        if not pid:
            continue
        row = conn.execute(
            "SELECT embedding_json FROM pattern_embeddings WHERE pattern_id=?", (pid,)
        ).fetchone()
        if row:
            try:
                vec = _json.loads(row["embedding_json"])
                sim = _cosine_similarity(cand_vec, vec)
                if sim > max_sim:
                    max_sim = sim
            except Exception:
                pass
    conn.close()
    return max_sim


def get_all_pattern_embeddings_summary() -> List[Dict]:
    """Liste des patterns avec embedding disponible (pour l'Analytics Dashboard)."""
    conn = get_conn()
    rows = conn.execute("""
        SELECT pe.pattern_id, pe.model_version, pe.updated_at, cp.name, cp.asset_class
        FROM pattern_embeddings pe
        LEFT JOIN custom_patterns cp ON cp.id = pe.pattern_id
        ORDER BY pe.updated_at DESC
    """).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def _build_risk_recommendation(
    saturated: List[str],
    div_score: float,
    exp_dd: float,
    alerts: List[Dict],
    high_corr: List[Dict],
) -> Dict:
    messages = []
    level = "ok"

    if saturated:
        level = "danger"
        messages.append(
            f"Portefeuille sur-concentré sur les facteurs {', '.join(saturated)}. "
            "Les 2 prochains trades devraient cibler d'autres facteurs de risque."
        )
    if div_score < 40:
        level = max(level, "warning") if level == "ok" else level
        messages.append(
            f"Score de diversification faible ({div_score}%). "
            "Envisager des positions sur des classes d'actifs non corrélées."
        )
    if exp_dd > 25:
        level = "danger"
        messages.append(
            f"Drawdown attendu estimé à {exp_dd}%. "
            "Réduire l'exposition aux facteurs à haute volatilité."
        )
    if high_corr:
        pair = high_corr[0]
        level = max(level, "warning") if level == "ok" else level
        messages.append(
            f"Attention : '{pair['name_a']}' et '{pair['name_b']}' sont fortement corrélés "
            f"({pair['correlation']:+.2f}) — ils ne représentent pas deux opportunités indépendantes."
        )

    if not messages:
        messages.append(
            "Portefeuille bien diversifié. "
            "Continuer à varier les facteurs de risque à chaque nouveau trade."
        )

    return {
        "level": level,
        "messages": messages,
        "summary": messages[0] if messages else "",
    }


# ── Cycle reports ─────────────────────────────────────────────────────────────

def save_cycle_report(run_id: str, report: Dict[str, Any]) -> None:
    import json as _json
    conn = get_conn()
    conn.execute(
        """INSERT OR REPLACE INTO cycle_reports
           (run_id, generated_at, macro_dominant, geo_score,
            patterns_added, trades_logged, trades_closed,
            pnl_snapshot_id, var_snapshot_id, full_report_json)
           VALUES (?, datetime('now'), ?, ?, ?, ?, ?, ?, ?, ?)""",
        (
            run_id,
            report.get("macro_dominant"),
            report.get("geo_score"),
            report.get("patterns_added", 0),
            report.get("trades_logged", 0),
            report.get("trades_closed", 0),
            report.get("pnl_snapshot_id"),
            report.get("var_snapshot_id"),
            _json.dumps(report, ensure_ascii=False),
        ),
    )
    conn.commit()
    conn.close()


def get_cycle_reports(limit: int = 20) -> List[Dict[str, Any]]:
    conn = get_conn()
    rows = conn.execute(
        """SELECT run_id, generated_at, macro_dominant, geo_score,
                  patterns_added, trades_logged, trades_closed
           FROM cycle_reports ORDER BY generated_at DESC LIMIT ?""",
        (limit,),
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_cycle_report(run_id: str) -> Optional[Dict[str, Any]]:
    import json as _json
    conn = get_conn()
    row = conn.execute(
        "SELECT full_report_json, generated_at FROM cycle_reports WHERE run_id=?",
        (run_id,),
    ).fetchone()
    conn.close()
    if not row or not row["full_report_json"]:
        return None
    try:
        report = _json.loads(row["full_report_json"])
        report["generated_at"] = row["generated_at"]
        return report
    except Exception:
        return None


def get_latest_cycle_report() -> Optional[Dict[str, Any]]:
    import json as _json
    conn = get_conn()
    row = conn.execute(
        "SELECT full_report_json, generated_at FROM cycle_reports ORDER BY generated_at DESC LIMIT 1"
    ).fetchone()
    conn.close()
    if not row or not row["full_report_json"]:
        return None
    try:
        report = _json.loads(row["full_report_json"])
        report["generated_at"] = row["generated_at"]
        return report
    except Exception:
        return None


def get_trade_assessments(run_id: str) -> List[Dict[str, Any]]:
    import json as _json
    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM options_trade_assessments WHERE run_id=? ORDER BY id",
        (run_id,),
    ).fetchall()
    conn.close()
    result = []
    for r in rows:
        d = dict(r)
        try:
            d["issues"] = _json.loads(d.get("issues_json") or "[]")
        except Exception:
            d["issues"] = []
        result.append(d)
    return result


def get_latest_trade_assessments(limit: int = 20) -> List[Dict[str, Any]]:
    """Return most recent assessments (one per trade_id) — for Journal badges."""
    import json as _json
    conn = get_conn()
    rows = conn.execute(
        """SELECT a.* FROM options_trade_assessments a
           INNER JOIN (SELECT trade_id, MAX(id) as max_id FROM options_trade_assessments
                       WHERE trade_id IS NOT NULL GROUP BY trade_id) m
           ON a.id = m.max_id
           ORDER BY a.id DESC LIMIT ?""",
        (limit,),
    ).fetchall()
    conn.close()
    result = []
    for r in rows:
        d = dict(r)
        try:
            d["issues"] = _json.loads(d.get("issues_json") or "[]")
        except Exception:
            d["issues"] = []
        result.append(d)
    return result


# ── Economic events (Phase 2) ─────────────────────────────────────────────────

def save_economic_event(
    event_name: str,
    series_id: str,
    event_date: str,
    actual_value: Optional[float],
    actual_unit: str = "",
    forecast_value: Optional[float] = None,
    previous_value: Optional[float] = None,
    surprise_pct: Optional[float] = None,
    surprise_zscore: Optional[float] = None,
    surprise_direction: str = "neutral",
    assets_impacted: Optional[List] = None,
    source: str = "FRED",
) -> Optional[int]:
    conn = get_conn()
    try:
        conn.execute(
            """INSERT INTO economic_events
               (event_name, series_id, event_date, actual_value, actual_unit,
                forecast_value, previous_value, surprise_pct, surprise_zscore,
                surprise_direction, assets_impacted, source)
               VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
               ON CONFLICT(series_id, event_date) DO UPDATE SET
                 actual_value=excluded.actual_value,
                 forecast_value=excluded.forecast_value,
                 previous_value=excluded.previous_value,
                 surprise_pct=excluded.surprise_pct,
                 surprise_zscore=excluded.surprise_zscore,
                 surprise_direction=excluded.surprise_direction,
                 fetched_at=datetime('now')""",
            (
                event_name, series_id, event_date, actual_value, actual_unit,
                forecast_value, previous_value, surprise_pct, surprise_zscore,
                surprise_direction,
                json.dumps(assets_impacted or []),
                source,
            ),
        )
        conn.commit()
        return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    finally:
        conn.close()


def get_recent_economic_surprises(days: int = 30, min_zscore: float = 0.0) -> List[Dict[str, Any]]:
    conn = get_conn()
    try:
        cutoff = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%d")
        rows = conn.execute(
            """SELECT * FROM economic_events
               WHERE event_date >= ?
                 AND ABS(COALESCE(surprise_zscore, 0)) >= ?
               ORDER BY event_date DESC, ABS(COALESCE(surprise_zscore, 0)) DESC
               LIMIT 20""",
            (cutoff, min_zscore),
        ).fetchall()
        result = []
        for r in rows:
            d = dict(r)
            try:
                d["assets_impacted"] = json.loads(d.get("assets_impacted") or "[]")
            except Exception:
                d["assets_impacted"] = []
            result.append(d)
        return result
    finally:
        conn.close()


def get_economic_events_for_calendar(days_back: int = 30, days_forward: int = 14) -> List[Dict[str, Any]]:
    """Return stored economic events for calendar enrichment."""
    conn = get_conn()
    try:
        cutoff_past = (datetime.utcnow() - timedelta(days=days_back)).strftime("%Y-%m-%d")
        cutoff_future = (datetime.utcnow() + timedelta(days=days_forward)).strftime("%Y-%m-%d")
        rows = conn.execute(
            """SELECT * FROM economic_events
               WHERE event_date BETWEEN ? AND ?
               ORDER BY event_date DESC""",
            (cutoff_past, cutoff_future),
        ).fetchall()
        result = []
        for r in rows:
            d = dict(r)
            try:
                d["assets_impacted"] = json.loads(d.get("assets_impacted") or "[]")
            except Exception:
                d["assets_impacted"] = []
            result.append(d)
        return result
    finally:
        conn.close()