performance_tracker.py

"""
performance_tracker.py
Logs every trade WickBot takes (demo or real) to a local SQLite database,
and computes the stats used to decide whether the demo "training" account
has earned the right to be considered for real trading at all.

Graduation is necessary but NOT sufficient for real trading — see
config.ALLOW_REAL_TRADING and telegram_bot.py's explicit confirmation
flow. This module only reports the numbers; it never flips the switch
by itself.
"""
import sqlite3
from contextlib import closing
from dataclasses import dataclass
from datetime import datetime

from config import Config

SCHEMA = """
CREATE TABLE IF NOT EXISTS trades (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    opened_at TEXT,
    closed_at TEXT,
    account_type TEXT,       -- 'demo' or 'real'
    symbol TEXT,
    strategy TEXT,
    pattern TEXT,
    side TEXT,
    score INTEGER,
    grade TEXT,
    entry REAL,
    sl REAL,                 -- current/last-known SL (position_manager.py updates this as it moves)
    tp REAL,
    volume REAL,
    result_r REAL,           -- realized R-multiple, filled in on close
    status TEXT,             -- 'open', 'closed_win', 'closed_loss', 'closed_be'
    ticket INTEGER,          -- MT5 position ticket, used to match open positions for management
    initial_sl REAL,         -- SL at the moment the trade was opened — the fixed risk basis for R calculations
    breakeven_moved INTEGER DEFAULT 0,
    trailing_active INTEGER DEFAULT 0
)
"""

# Columns added after the original schema — applied defensively so
# existing databases from before this feature still work without
# needing a manual migration step.
_MIGRATION_COLUMNS = [
    ("ticket", "INTEGER"),
    ("initial_sl", "REAL"),
    ("breakeven_moved", "INTEGER DEFAULT 0"),
    ("trailing_active", "INTEGER DEFAULT 0"),
    ("risk_amount", "REAL"),  # $ risked at open — the denominator for realized result_r on close
    ("exit_reason", "TEXT"),  # 'tp' / 'sl' / 'other' — classified in position_manager.reconcile_closed_trades()
                              # by comparing the actual exit price to the trade's stored tp/sl, so the
                              # auto-tuner can tell whether a strategy's TP/SL levels are well-calibrated
                              # (e.g. TP getting hit almost every time may mean it's set too conservatively)
]


def _connect():
    conn = sqlite3.connect(Config.DB_PATH)
    conn.execute(SCHEMA)
    for col_name, col_type in _MIGRATION_COLUMNS:
        try:
            conn.execute(f"ALTER TABLE trades ADD COLUMN {col_name} {col_type}")
        except sqlite3.OperationalError:
            pass  # column already exists
    return conn


def log_trade_open(account_type, symbol, strategy, pattern, side, score, grade, entry, sl, tp, volume,
                    ticket=None, risk_amount=None) -> int:
    with closing(_connect()) as conn:
        cur = conn.execute(
            """INSERT INTO trades
               (opened_at, account_type, symbol, strategy, pattern, side, score, grade,
                entry, sl, tp, volume, status, ticket, initial_sl, risk_amount)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?, ?)""",
            (
                datetime.utcnow().isoformat(), account_type, symbol, strategy, pattern,
                side, score, grade, entry, sl, tp, volume, ticket, sl, risk_amount,
            ),
        )
        conn.commit()
        return cur.lastrowid


def log_trade_close(trade_id: int, result_r: float, status: str, exit_reason: str = None):
    with closing(_connect()) as conn:
        conn.execute(
            "UPDATE trades SET closed_at=?, result_r=?, status=?, exit_reason=? WHERE id=?",
            (datetime.utcnow().isoformat(), result_r, status, exit_reason, trade_id),
        )
        conn.commit()


def get_exit_reason_stats(strategy: str) -> dict:
    """Counts of how closed trades for a given strategy actually exited
    ('tp' / 'sl' / 'other') — used by auto_tuner.py to check whether a
    strategy's own TP/SL calibration looks off (e.g. TP hit almost every
    time may mean it's set too close; SL hit almost every time with a
    low win rate may mean it's set too tight or the TP too far)."""
    with closing(_connect()) as conn:
        rows = conn.execute(
            "SELECT exit_reason, COUNT(*) FROM trades "
            "WHERE account_type='demo' AND status LIKE 'closed_%' AND strategy=? "
            "GROUP BY exit_reason",
            (strategy,),
        ).fetchall()
    counts = {reason or "other": count for reason, count in rows}
    total = sum(counts.values())
    return {"total": total, "counts": counts}


def get_open_trade_by_ticket(ticket: int) -> dict:
    """Used by position_manager.py to find a position's original entry
    and risk basis (initial_sl) — the fixed reference point for
    calculating profit in R, no matter how many times the live SL has
    since been moved."""
    with closing(_connect()) as conn:
        row = conn.execute(
            """SELECT id, symbol, side, entry, initial_sl, sl, tp, breakeven_moved,
                      trailing_active, opened_at, volume, risk_amount
               FROM trades WHERE ticket=? AND status='open'""",
            (ticket,),
        ).fetchone()
    if not row:
        return None
    return {
        "id": row[0], "symbol": row[1], "side": row[2], "entry": row[3],
        "initial_sl": row[4], "current_sl": row[5], "tp": row[6],
        "breakeven_moved": bool(row[7]), "trailing_active": bool(row[8]),
        "opened_at": row[9], "volume": row[10], "risk_amount": row[11],
    }


def get_all_open_trades(ticket_not_null: bool = True) -> list:
    """Used by position_manager.py's closed-trade reconciliation — every
    trade the DB still thinks is 'open', so it can check which of those
    have actually closed on the MT5 side since the last check."""
    query = ("SELECT id, ticket, symbol, side, entry, initial_sl, tp, risk_amount, "
             "account_type, strategy FROM trades WHERE status='open'")
    if ticket_not_null:
        query += " AND ticket IS NOT NULL"
    with closing(_connect()) as conn:
        rows = conn.execute(query).fetchall()
    return [
        {
            "id": r[0], "ticket": r[1], "symbol": r[2], "side": r[3],
            "entry": r[4], "initial_sl": r[5], "tp": r[6], "risk_amount": r[7],
            "account_type": r[8], "strategy": r[9],
        }
        for r in rows
    ]


def update_trade_sl(trade_id: int, new_sl: float, breakeven_moved: bool = None, trailing_active: bool = None):
    """Records a stop-loss move made by position_manager.py against the
    trade log, purely for record-keeping — this never affects the fixed
    initial_sl risk basis used for R calculations."""
    sets = ["sl=?"]
    params = [new_sl]
    if breakeven_moved is not None:
        sets.append("breakeven_moved=?")
        params.append(1 if breakeven_moved else 0)
    if trailing_active is not None:
        sets.append("trailing_active=?")
        params.append(1 if trailing_active else 0)
    params.append(trade_id)
    with closing(_connect()) as conn:
        conn.execute(f"UPDATE trades SET {', '.join(sets)} WHERE id=?", params)
        conn.commit()


@dataclass
class DemoStats:
    total_closed: int
    win_rate: float
    expectancy_r: float
    by_strategy: dict
    by_symbol: dict
    by_strategy_symbol: dict


def _agg(rows_with_key):
    """rows_with_key: iterable of (key, result_r). Returns {key: {trades, win_rate, expectancy_r}}.
    
    CRITICAL: result_r can be 0.0 (breakeven trade). Using `result_r or 0.0` would
    incorrectly replace 0.0 with 0.0 (no-op in Python since 0.0 is falsy), but the
    intent is to handle None values. Use `result_r if result_r is not None else 0.0`
    to preserve actual 0.0 values while still handling None."""
    buckets = {}
    for key, result_r in rows_with_key:
        buckets.setdefault(key, []).append(result_r if result_r is not None else 0.0)
    return {
        key: {
            "trades": len(vals),
            "win_rate": round(sum(1 for v in vals if v > 0) / len(vals), 2),
            "expectancy_r": round(sum(vals) / len(vals), 2),
        }
        for key, vals in buckets.items()
    }


def get_stats_by_grade() -> dict:
    """Used by auto_tuner.py to check whether lower-graded signals
    (C/D) are actually net-negative, as evidence for a MIN_SIGNAL_SCORE
    suggestion — never used to determine the suggested value directly,
    just to check whether raising the threshold is statistically
    supported at all."""
    with closing(_connect()) as conn:
        rows = conn.execute(
            "SELECT grade, result_r FROM trades WHERE account_type='demo' AND status LIKE 'closed_%'"
        ).fetchall()
    return _agg((grade, result_r) for grade, result_r in rows)


def get_demo_stats() -> DemoStats:
    with closing(_connect()) as conn:
        rows = conn.execute(
            "SELECT strategy, symbol, result_r, status FROM trades "
            "WHERE account_type='demo' AND status LIKE 'closed_%'"
        ).fetchall()

    if not rows:
        return DemoStats(0, 0.0, 0.0, {}, {}, {})

    total = len(rows)
    wins = sum(1 for r in rows if r[2] is not None and r[2] > 0)
    win_rate = wins / total
    expectancy = sum(r[2] for r in rows if r[2] is not None) / total

    by_strategy = _agg((strat, result_r) for strat, symbol, result_r, _ in rows)
    by_symbol = _agg((symbol, result_r) for strat, symbol, result_r, _ in rows)
    by_strategy_symbol = _agg((f"{strat} / {symbol}", result_r) for strat, symbol, result_r, _ in rows)

    return DemoStats(
        total, round(win_rate, 2), round(expectancy, 2),
        by_strategy, by_symbol, by_strategy_symbol,
    )


def graduation_status() -> dict:
    """
    Returns whether the demo track record meets the configured bar for
    even being *eligible* to consider real trading. This is a self-imposed
    training gate, not a guarantee of future performance.
    """
    stats = get_demo_stats()
    meets_trades = stats.total_closed >= Config.MIN_DEMO_TRADES
    meets_winrate = stats.win_rate >= Config.MIN_DEMO_WIN_RATE
    meets_expectancy = stats.expectancy_r >= Config.MIN_DEMO_EXPECTANCY_R
    eligible = meets_trades and meets_winrate and meets_expectancy
    return {
        "eligible": eligible,
        "stats": stats,
        "requirements": {
            "trades": (stats.total_closed, Config.MIN_DEMO_TRADES, meets_trades),
            "win_rate": (stats.win_rate, Config.MIN_DEMO_WIN_RATE, meets_winrate),
            "expectancy_r": (stats.expectancy_r, Config.MIN_DEMO_EXPECTANCY_R, meets_expectancy),
        },
    }