database/db.py

"""
SQLite database layer — async via aiosqlite.
Tables: agents, agent_history, news, predictions, trades, debate_cycles.
"""

import aiosqlite
import json
import logging
from datetime import datetime
from pathlib import Path

logger = logging.getLogger("gap_system.db")

DB_PATH: Path | None = None


async def init_db(db_path: Path) -> None:
    """Create all tables if they don't exist."""
    global DB_PATH
    DB_PATH = db_path
    db_path.parent.mkdir(parents=True, exist_ok=True)

    async with aiosqlite.connect(str(db_path)) as db:
        await db.executescript("""
            CREATE TABLE IF NOT EXISTS agents (
                id            TEXT PRIMARY KEY,
                platform      TEXT NOT NULL,
                name          TEXT NOT NULL,
                personality   TEXT NOT NULL,
                memory        TEXT DEFAULT '[]',
                stance        TEXT DEFAULT 'NEUTRAL',
                confidence    REAL DEFAULT 0.5,
                reasoning     TEXT DEFAULT '',
                influence     REAL DEFAULT 1.0,
                total_predictions INTEGER DEFAULT 0,
                correct_predictions INTEGER DEFAULT 0,
                consecutive_failures INTEGER DEFAULT 0,
                created_at    TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS agent_history (
                rowid         INTEGER PRIMARY KEY AUTOINCREMENT,
                agent_id      TEXT NOT NULL,
                cycle_id      TEXT NOT NULL,
                stance        TEXT NOT NULL,
                confidence    REAL NOT NULL,
                reasoning     TEXT NOT NULL,
                influenced_by TEXT DEFAULT '[]',
                key_concern   TEXT DEFAULT '',
                timestamp     TEXT DEFAULT (datetime('now')),
                FOREIGN KEY (agent_id) REFERENCES agents(id)
            );

            CREATE TABLE IF NOT EXISTS news (
                rowid         INTEGER PRIMARY KEY AUTOINCREMENT,
                source        TEXT NOT NULL,
                title         TEXT NOT NULL,
                summary       TEXT DEFAULT '',
                sentiment     TEXT DEFAULT 'neutral',
                importance    TEXT DEFAULT 'medium',
                category      TEXT DEFAULT '',
                asset         TEXT DEFAULT '',
                url           TEXT DEFAULT '',
                published_at  TEXT DEFAULT '',
                ingested_at   TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS predictions (
                rowid         INTEGER PRIMARY KEY AUTOINCREMENT,
                asset         TEXT NOT NULL,
                direction     TEXT NOT NULL,
                confidence    REAL NOT NULL,
                lot_size      REAL,
                hold_seconds  INTEGER,
                strategy      TEXT DEFAULT '',
                locked        INTEGER DEFAULT 0,
                week_label    TEXT NOT NULL,
                created_at    TEXT DEFAULT (datetime('now')),
                updated_at    TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS trades (
                rowid         INTEGER PRIMARY KEY AUTOINCREMENT,
                asset         TEXT NOT NULL,
                direction     TEXT NOT NULL,
                lot_size      REAL NOT NULL,
                hold_seconds  INTEGER NOT NULL,
                strategy      TEXT DEFAULT '',
                entry_price   REAL,
                exit_price    REAL,
                profit_pips   REAL,
                profit_usd    REAL,
                status        TEXT DEFAULT 'pending',
                error         TEXT DEFAULT '',
                executed_at   TEXT DEFAULT (datetime('now')),
                closed_at     TEXT DEFAULT ''
            );

            CREATE TABLE IF NOT EXISTS debate_cycles (
                rowid         INTEGER PRIMARY KEY AUTOINCREMENT,
                cycle_id      TEXT NOT NULL UNIQUE,
                cycle_type    TEXT NOT NULL,
                week_label    TEXT NOT NULL,
                agents_run    INTEGER DEFAULT 0,
                agents_failed INTEGER DEFAULT 0,
                bull_votes    INTEGER DEFAULT 0,
                bear_votes    INTEGER DEFAULT 0,
                neutral_votes INTEGER DEFAULT 0,
                started_at    TEXT DEFAULT (datetime('now')),
                finished_at   TEXT DEFAULT ''
            );

            CREATE TABLE IF NOT EXISTS economic_data (
                rowid         INTEGER PRIMARY KEY AUTOINCREMENT,
                series_id     TEXT NOT NULL,
                series_name   TEXT NOT NULL,
                value         REAL,
                previous      REAL,
                deviation     REAL,
                fetched_at    TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS cot_data (
                rowid         INTEGER PRIMARY KEY AUTOINCREMENT,
                asset         TEXT NOT NULL,
                net_long      REAL,
                net_short     REAL,
                net_position  REAL,
                week_change   REAL,
                extreme_flag  INTEGER DEFAULT 0,
                bias          TEXT DEFAULT 'NEUTRAL',
                report_date   TEXT DEFAULT '',
                fetched_at    TEXT DEFAULT (datetime('now'))
            );
        """)
        await db.execute("PRAGMA journal_mode=WAL")
        await db.execute("PRAGMA busy_timeout=5000")
        
        # Apply schema migrations to existing databases
        migrations = [
            "ALTER TABLE agents ADD COLUMN total_predictions INTEGER DEFAULT 0;",
            "ALTER TABLE agents ADD COLUMN correct_predictions INTEGER DEFAULT 0;",
            "ALTER TABLE agents ADD COLUMN consecutive_failures INTEGER DEFAULT 0;",
            "ALTER TABLE debate_cycles ADD COLUMN bull_votes INTEGER DEFAULT 0;",
            "ALTER TABLE debate_cycles ADD COLUMN bear_votes INTEGER DEFAULT 0;",
            "ALTER TABLE debate_cycles ADD COLUMN neutral_votes INTEGER DEFAULT 0;",
        ]
        for query in migrations:
            try:
                await db.execute(query)
            except aiosqlite.OperationalError:
                pass  # duplicate column
        
        # Migrations for expanded historical_gaps features (quant model v2)
        gap_migrations = [
            "ALTER TABLE historical_gaps ADD COLUMN vix_change_5d REAL DEFAULT 0.0;",
            "ALTER TABLE historical_gaps ADD COLUMN weekly_return REAL DEFAULT 0.0;",
            "ALTER TABLE historical_gaps ADD COLUMN rsi_14 REAL DEFAULT 50.0;",
            "ALTER TABLE historical_gaps ADD COLUMN macd_hist REAL DEFAULT 0.0;",
            "ALTER TABLE historical_gaps ADD COLUMN ema_spread REAL DEFAULT 0.0;",
            "ALTER TABLE historical_gaps ADD COLUMN bb_width REAL DEFAULT 0.02;",
            "ALTER TABLE historical_gaps ADD COLUMN dxy_change REAL DEFAULT 0.0;",
            "ALTER TABLE historical_gaps ADD COLUMN prev_gap_dir TEXT DEFAULT 'NONE';",
        ]
        for query in gap_migrations:
            try:
                await db.execute(query)
            except aiosqlite.OperationalError:
                pass  # column already exists
        
        # v3 migrations: expanded gap features
        gap_v3_migrations = [
            "ALTER TABLE historical_gaps ADD COLUMN atr_14 REAL DEFAULT 1.0;",
            "ALTER TABLE historical_gaps ADD COLUMN gap_atr_ratio REAL DEFAULT 0.0;",
            "ALTER TABLE historical_gaps ADD COLUMN day_of_month INTEGER DEFAULT 15;",
            "ALTER TABLE historical_gaps ADD COLUMN prev_3_gaps_mean REAL DEFAULT 0.0;",
            "ALTER TABLE historical_gaps ADD COLUMN gap_fill_rate REAL DEFAULT 0.6;",
            "ALTER TABLE historical_gaps ADD COLUMN gold_dxy_ratio REAL DEFAULT 0.0;",
            "ALTER TABLE historical_gaps ADD COLUMN gld_momentum REAL DEFAULT 0.0;",
            "ALTER TABLE historical_gaps ADD COLUMN real_yield_proxy REAL DEFAULT 0.0;",
            "ALTER TABLE historical_gaps ADD COLUMN treasury_10y REAL DEFAULT 3.0;",
        ]
        for query in gap_v3_migrations:
            try:
                await db.execute(query)
            except aiosqlite.OperationalError:
                pass  # column already exists
        
        # New tables for Bayesian engine and economic calendar
        await db.executescript("""
            CREATE TABLE IF NOT EXISTS bayesian_history (
                rowid           INTEGER PRIMARY KEY AUTOINCREMENT,
                asset           TEXT NOT NULL,
                week_label      TEXT NOT NULL,
                cycle_number    INTEGER DEFAULT 0,
                p_bullish       REAL NOT NULL,
                p_bearish       REAL NOT NULL,
                log_odds        REAL NOT NULL,
                evidence_count  INTEGER DEFAULT 0,
                evidence_list   TEXT DEFAULT '[]',
                direction       TEXT NOT NULL,
                confidence      REAL NOT NULL,
                timestamp       TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS economic_events (
                rowid           INTEGER PRIMARY KEY AUTOINCREMENT,
                event_id        TEXT NOT NULL,
                event_name      TEXT NOT NULL,
                country         TEXT NOT NULL,
                impact          TEXT NOT NULL,
                release_date    TEXT NOT NULL,
                forecast_value  REAL,
                actual_value    REAL,
                previous_value  REAL,
                surprise        REAL,
                surprise_pct    REAL,
                status          TEXT DEFAULT 'PENDING',
                created_at      TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS historical_gaps (
                rowid           INTEGER PRIMARY KEY AUTOINCREMENT,
                asset           TEXT NOT NULL,
                friday_date     TEXT NOT NULL,
                friday_close    REAL NOT NULL,
                monday_open     REAL NOT NULL,
                gap_size        REAL NOT NULL,
                gap_pct         REAL NOT NULL,
                gap_direction   TEXT NOT NULL,
                vix_close       REAL,
                volume_spike    REAL,
                weekly_trend    TEXT
            );
        """)
        
        await db.commit()
    logger.info("Database initialised at %s (WAL mode)", db_path)


def _conn():
    """Return a context manager for connecting."""
    if DB_PATH is None:
        raise RuntimeError("Database not initialised — call init_db() first")
    return aiosqlite.connect(str(DB_PATH))


# ── Agent CRUD ───────────────────────────────────────────────────────────────

async def upsert_agent(agent_id: str, platform: str, name: str,
                       personality: str, influence: float = 1.0) -> None:
    async with _conn() as db:
        await db.execute(
            """INSERT INTO agents (id, platform, name, personality, influence)
               VALUES (?, ?, ?, ?, ?)
               ON CONFLICT(id) DO UPDATE SET
                   platform=excluded.platform,
                   name=excluded.name,
                   personality=excluded.personality
                   -- do not overwrite influence and performance scores on restart
                   """,
            (agent_id, platform, name, personality, influence),
        )
        await db.commit()


async def update_agent_performance(agent_id: str, is_correct: bool) -> None:
    inc = 1 if is_correct else 0
    async with _conn() as db:
        await db.execute(
            """UPDATE agents SET 
               total_predictions = total_predictions + 1,
               correct_predictions = correct_predictions + ?
               WHERE id=?""",
            (inc, agent_id),
        )
        await db.commit()


async def record_agent_failure(agent_id: str) -> int:
    """Increments consecutive failures and returns new count."""
    async with _conn() as db:
        await db.execute(
            "UPDATE agents SET consecutive_failures = consecutive_failures + 1 WHERE id=?",
            (agent_id,)
        )
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT consecutive_failures FROM agents WHERE id=?", (agent_id,))
        row = await cursor.fetchone()
        await db.commit()
        return row["consecutive_failures"] if row else 0


async def reset_agent_failure(agent_id: str) -> None:
    async with _conn() as db:
        await db.execute(
            "UPDATE agents SET consecutive_failures = 0 WHERE id=?",
            (agent_id,)
        )
        await db.commit()


async def update_agent_state(agent_id: str, stance: str, confidence: float,
                             reasoning: str, memory: list[str]) -> None:
    async with _conn() as db:
        await db.execute(
            """UPDATE agents SET stance=?, confidence=?, reasoning=?,
               memory=?, influence=influence WHERE id=?""",
            (stance, confidence, reasoning, json.dumps(memory), agent_id),
        )
        await db.commit()


async def get_all_agents() -> list[dict]:
    async with _conn() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT * FROM agents ORDER BY id")
        rows = await cursor.fetchall()
        return [dict(r) for r in rows]


# ── Agent History ────────────────────────────────────────────────────────────

async def log_agent_cycle(agent_id: str, cycle_id: str, stance: str,
                          confidence: float, reasoning: str,
                          influenced_by: list[str],
                          key_concern: str = "") -> None:
    async with _conn() as db:
        await db.execute(
            """INSERT INTO agent_history
               (agent_id, cycle_id, stance, confidence, reasoning,
                influenced_by, key_concern)
               VALUES (?, ?, ?, ?, ?, ?, ?)""",
            (agent_id, cycle_id, stance, confidence, reasoning,
             json.dumps(influenced_by), key_concern),
        )
        await db.commit()


async def get_agent_history(agent_id: str, limit: int = 20) -> list[dict]:
    async with _conn() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(
            """SELECT * FROM agent_history
               WHERE agent_id=? ORDER BY rowid DESC LIMIT ?""",
            (agent_id, limit),
        )
        return [dict(r) for r in await cursor.fetchall()]


# ── News ─────────────────────────────────────────────────────────────────────

async def insert_news(source: str, title: str, summary: str = "",
                      sentiment: str = "neutral", importance: str = "medium",
                      category: str = "", asset: str = "",
                      url: str = "", published_at: str = "") -> None:
    async with _conn() as db:
        await db.execute(
            """INSERT INTO news
               (source, title, summary, sentiment, importance, category,
                asset, url, published_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (source, title, summary, sentiment, importance, category,
             asset, url, published_at),
        )
        await db.commit()


async def get_recent_news(limit: int = 50) -> list[dict]:
    async with _conn() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(
            "SELECT * FROM news ORDER BY rowid DESC LIMIT ?", (limit,))
        return [dict(r) for r in await cursor.fetchall()]


# ── Predictions ──────────────────────────────────────────────────────────────

async def upsert_prediction(asset: str, direction: str, confidence: float,
                            lot_size: float | None, hold_seconds: int | None,
                            strategy: str, locked: bool,
                            week_label: str) -> None:
    async with _conn() as db:
        # Check if a prediction already exists for this asset+week
        cursor = await db.execute(
            "SELECT rowid FROM predictions WHERE asset=? AND week_label=? ORDER BY rowid DESC LIMIT 1",
            (asset, week_label),
        )
        existing = await cursor.fetchone()
        
        if existing:
            # Update existing prediction
            await db.execute(
                """UPDATE predictions SET
                   direction=?, confidence=?, lot_size=?, hold_seconds=?,
                   strategy=?, locked=?, updated_at=datetime('now')
                   WHERE rowid=?""",
                (direction, confidence, lot_size, hold_seconds,
                 strategy, int(locked), existing[0]),
            )
        else:
            # Insert new prediction
            await db.execute(
                """INSERT INTO predictions
                   (asset, direction, confidence, lot_size, hold_seconds,
                    strategy, locked, week_label)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                (asset, direction, confidence, lot_size, hold_seconds,
                 strategy, int(locked), week_label),
            )
        await db.commit()


async def get_latest_prediction(asset: str) -> dict | None:
    async with _conn() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(
            """SELECT * FROM predictions
               WHERE asset=? ORDER BY rowid DESC LIMIT 1""",
            (asset,),
        )
        row = await cursor.fetchone()
        return dict(row) if row else None


async def get_all_predictions() -> list[dict]:
    async with _conn() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(
            "SELECT * FROM predictions ORDER BY rowid DESC")
        return [dict(r) for r in await cursor.fetchall()]


# ── Trades ───────────────────────────────────────────────────────────────────

async def insert_trade(asset: str, direction: str, lot_size: float,
                       hold_seconds: int, strategy: str = "",
                       entry_price: float = 0.0) -> int:
    async with _conn() as db:
        cursor = await db.execute(
            """INSERT INTO trades
               (asset, direction, lot_size, hold_seconds, strategy,
                entry_price, status)
               VALUES (?, ?, ?, ?, ?, ?, 'open')""",
            (asset, direction, lot_size, hold_seconds, strategy, entry_price),
        )
        await db.commit()
        return cursor.lastrowid  # type: ignore[return-value]


async def close_trade(trade_id: int, exit_price: float,
                      profit_pips: float, profit_usd: float) -> None:
    async with _conn() as db:
        await db.execute(
            """UPDATE trades SET exit_price=?, profit_pips=?, profit_usd=?,
               status='closed', closed_at=datetime('now') WHERE rowid=?""",
            (exit_price, profit_pips, profit_usd, trade_id),
        )
        await db.commit()


async def fail_trade(trade_id: int, error: str) -> None:
    async with _conn() as db:
        await db.execute(
            "UPDATE trades SET status='failed', error=? WHERE rowid=?",
            (error, trade_id),
        )
        await db.commit()


async def get_trade_history(limit: int = 100) -> list[dict]:
    async with _conn() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(
            "SELECT * FROM trades ORDER BY rowid DESC LIMIT ?", (limit,))
        return [dict(r) for r in await cursor.fetchall()]


# ── Debate Cycles ────────────────────────────────────────────────────────────

async def start_debate_cycle(cycle_id: str, cycle_type: str,
                             week_label: str) -> None:
    async with _conn() as db:
        await db.execute(
            """INSERT INTO debate_cycles (cycle_id, cycle_type, week_label)
               VALUES (?, ?, ?)""",
            (cycle_id, cycle_type, week_label),
        )
        await db.commit()


async def finish_debate_cycle(cycle_id: str, agents_run: int,
                              agents_failed: int) -> None:
    async with _conn() as db:
        await db.execute(
            """UPDATE debate_cycles
               SET agents_run=?, agents_failed=?, finished_at=datetime('now')
               WHERE cycle_id=?""",
            (agents_run, agents_failed, cycle_id),
        )
        await db.commit()


# ── Economic Data ────────────────────────────────────────────────────────────

async def insert_economic(series_id: str, series_name: str,
                          value: float, previous: float,
                          deviation: float) -> None:
    async with _conn() as db:
        await db.execute(
            """INSERT INTO economic_data
               (series_id, series_name, value, previous, deviation)
               VALUES (?, ?, ?, ?, ?)""",
            (series_id, series_name, value, previous, deviation),
        )
        await db.commit()


# ── COT Data ────────────────────────────────────────────────────────────────

async def insert_cot(asset: str, net_long: float, net_short: float,
                     net_position: float, week_change: float,
                     extreme_flag: bool, bias: str,
                     report_date: str) -> None:
    async with _conn() as db:
        await db.execute(
            """INSERT INTO cot_data
               (asset, net_long, net_short, net_position, week_change,
                extreme_flag, bias, report_date)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
            (asset, net_long, net_short, net_position, week_change,
             int(extreme_flag), bias, report_date),
        )
        await db.commit()


async def get_latest_cot(asset: str) -> dict | None:
    async with _conn() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(
            "SELECT * FROM cot_data WHERE asset=? ORDER BY rowid DESC LIMIT 1",
            (asset,),
        )
        row = await cursor.fetchone()
        return dict(row) if row else None


# ── Bayesian History ─────────────────────────────────────────────────────────

async def log_bayesian_update(
    asset: str, week_label: str, cycle_number: int,
    p_bullish: float, p_bearish: float, log_odds: float,
    evidence_count: int, evidence_list: list[str],
    direction: str, confidence: float,
) -> None:
    """Log a Bayesian posterior update for audit trail."""
    async with _conn() as db:
        await db.execute(
            """INSERT INTO bayesian_history
               (asset, week_label, cycle_number, p_bullish, p_bearish,
                log_odds, evidence_count, evidence_list, direction, confidence)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (asset, week_label, cycle_number, p_bullish, p_bearish,
             log_odds, evidence_count, json.dumps(evidence_list),
             direction, confidence),
        )
        await db.commit()


async def get_bayesian_history(asset: str, limit: int = 50) -> list[dict]:
    """Get recent Bayesian update history for an asset."""
    async with _conn() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(
            "SELECT * FROM bayesian_history WHERE asset=? ORDER BY rowid DESC LIMIT ?",
            (asset, limit),
        )
        rows = await cursor.fetchall()
        return [dict(r) for r in rows]


# ── Economic Events ──────────────────────────────────────────────────────────

async def log_economic_event(
    event_id: str, event_name: str, country: str, impact: str,
    release_date: str, forecast_value: float | None,
    actual_value: float | None, previous_value: float | None,
    surprise: float | None, surprise_pct: float | None,
    status: str,
) -> None:
    """Log an economic event release for historical analysis."""
    async with _conn() as db:
        await db.execute(
            """INSERT INTO economic_events
               (event_id, event_name, country, impact, release_date,
                forecast_value, actual_value, previous_value,
                surprise, surprise_pct, status)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (event_id, event_name, country, impact, release_date,
             forecast_value, actual_value, previous_value,
             surprise, surprise_pct, status),
        )
        await db.commit()


async def get_economic_events(limit: int = 30) -> list[dict]:
    """Get recent economic events with actual results."""
    async with _conn() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(
            "SELECT * FROM economic_events ORDER BY rowid DESC LIMIT ?",
            (limit,),
        )
        rows = await cursor.fetchall()
        return [dict(r) for r in rows]