Chan/ChanMacro/expectancy/engine.py

"""
expectancy/engine.py — Bayesian Expectancy Engine.

Core algorithm:
  1. LeveledExpectancy: filter layer-by-layer, stop at highest valid level
  2. Empirical Bayes prior: prior = signal's global historical winrate
  3. Dynamic Beta strength: adaptive to sample size
  4. Time decay: recent samples weighted higher (half_life=180d)
  5. SufficiencyGuard: refuse output if effective_samples < 30
  6. KNN Fallback: similarity search when strict filtering fails (Phase D)
"""

from datetime import date as Date
from typing import Optional
import sqlite3
import logging

import numpy as np
import pandas as pd

from models import (
    MarketStateVector, ExpectancyReport, ExpectancyLayer,
    SufficiencyLevel, MarketRegime,
)
from config import config
from .decay import TimeDecay

logger = logging.getLogger(__name__)


class SufficiencyGuard:
    """Prevents trading advice from insufficient samples."""

    def __init__(self, min_effective: int = 30, low: int = 50, medium: int = 100):
        self.MIN = min_effective
        self.LOW = low
        self.MEDIUM = medium

    def evaluate(self, effective_samples: float) -> SufficiencyLevel:
        if effective_samples < self.MIN:
            return SufficiencyLevel.INSUFFICIENT
        elif effective_samples < self.LOW:
            return SufficiencyLevel.LOW
        elif effective_samples < self.MEDIUM:
            return SufficiencyLevel.MEDIUM
        return SufficiencyLevel.HIGH


class BayesianExpectancyEngine:
    """
    Leveled Bayesian Expectancy Engine.

    Query layers from coarse to fine. Stop when effective_samples drops below threshold.
    Uses Empirical Bayes prior (signal's global winrate, not fixed 50%).
    """

    # Expectancy query levels: name → WHERE clause template
    LEVELS = [
        ("Base",            "signal_type = '{signal}'"),
        ("+ Regime",        "signal_type = '{signal}' AND regime = '{regime}'"),
        ("+ Breadth",       "signal_type = '{signal}' AND regime = '{regime}' AND breadth_bucket = '{breadth}'"),
        ("+ OI State",      "signal_type = '{signal}' AND regime = '{regime}' AND breadth_bucket = '{breadth}' AND oi_state = '{oi}'"),
        ("+ Volatility",    "signal_type = '{signal}' AND regime = '{regime}' AND breadth_bucket = '{breadth}' AND oi_state = '{oi}' AND volatility_regime = '{vol}'"),
    ]

    def __init__(self, db_path: Optional[str] = None,
                 half_life_days: int = 180,
                 level_min_samples: int = 50):
        self.db_path = db_path or config.db_path
        self.decay = TimeDecay(half_life_days)
        self.guard = SufficiencyGuard(
            min_effective=config.sufficiency_min_effective,
            low=config.sufficiency_low,
            medium=config.sufficiency_medium,
        )
        self.level_min = level_min_samples

    def estimate(self, state: MarketStateVector,
                 signal_type: str = "B3",
                 target_date: Optional[Date] = None) -> ExpectancyReport:
        """
        Compute layered Bayesian expectancy for a signal in current market state.

        Returns the estimate at the deepest level with >= level_min effective samples.
        """
        if target_date is None:
            target_date = Date.today()

        conn = sqlite3.connect(self.db_path)

        # Get global signal winrate for Empirical Bayes prior
        global_rate = self._global_winrate(conn, signal_type)

        layers = []
        best_result = None

        for level_name, template in self.LEVELS:
            where = template.format(
                signal=signal_type,
                regime=state.regime.value,
                breadth=state.breadth_bucket.value,
                oi=state.oi_state.value,
                vol=state.volatility_regime.value,
            )
            query = f"SELECT * FROM signal_features WHERE {where}"
            df = pd.read_sql_query(query, conn)

            if df.empty:
                layers.append(ExpectancyLayer(
                    name=level_name, posterior_winrate=0.0,
                    samples=0, effective_samples=0.0,
                ))
                continue

            # Time-weighted stats
            dates_list = [Date.fromisoformat(d) for d in df["date"]]
            weights = self.decay.weights(dates_list, target_date)
            eff_n = self.decay.effective_samples(weights)

            wins = pd.to_numeric(df["is_win_7d"].fillna(0), errors="coerce").fillna(0).values
            returns = pd.to_numeric(df["result_7d"].fillna(0), errors="coerce").fillna(0).values

            raw_wr = float(wins.mean()) if len(wins) > 0 else 0.0
            weighted_wr = self.decay.weighted_win_rate(wins, weights)
            weighted_ret = self.decay.weighted_mean(returns, weights)

            # Empirical Bayes posterior
            posterior = self._bayesian_posterior(
                global_rate=global_rate,
                wins=wins.sum(),
                samples=len(df),
            )

            layer = ExpectancyLayer(
                name=level_name,
                posterior_winrate=round(posterior, 4),
                raw_winrate=round(raw_wr, 4),
                samples=len(df),
                effective_samples=round(eff_n, 1),
                avg_return=round(weighted_ret, 2),
            )
            layers.append(layer)

            # Level-based fallback: keep going while samples sufficient
            if eff_n >= self.level_min:
                best_result = layer

        conn.close()

        if best_result is None and layers:
            # Fallback to the deepest layer that had any samples
            for layer in reversed(layers):
                if layer.samples > 0:
                    best_result = layer
                    break

        if best_result is None:
            return ExpectancyReport(
                signal_type=signal_type,
                date=target_date,
                layers=layers,
                final_estimate=0.0,
                sufficiency=SufficiencyLevel.INSUFFICIENT,
                source="insufficient",
            )

        sufficiency = self.guard.evaluate(
            best_result.effective_samples
        )

        # Compute profit factor and MAE from the SAME level as best_result
        profit_factor = None
        avg_mae = None
        if best_result and best_result.samples > 0:
            # Re-query the level that produced best_result
            best_level_idx = next(
                i for i, l in enumerate(layers) if l.name == best_result.name
            )
            where = self.LEVELS[best_level_idx][1].format(
                signal=signal_type, regime=state.regime.value,
                breadth=state.breadth_bucket.value, oi=state.oi_state.value,
                vol=state.volatility_regime.value,
            )
            query = f"SELECT result_7d, max_adverse_excursion FROM signal_features WHERE {where}"
            conn2 = sqlite3.connect(self.db_path)
            df_detail = pd.read_sql_query(query, conn2)
            conn2.close()
            if not df_detail.empty:
                returns_7d = df_detail["result_7d"].dropna()
                if len(returns_7d) > 0:
                    gains = returns_7d[returns_7d > 0].sum()
                    losses = abs(returns_7d[returns_7d < 0].sum())
                    profit_factor = round(gains / losses, 2) if losses > 0 else None
                maes = df_detail["max_adverse_excursion"].dropna()
                if len(maes) > 0:
                    avg_mae = round(float(maes.mean()), 2)

        return ExpectancyReport(
            signal_type=signal_type,
            date=target_date,
            layers=layers,
            final_estimate=round(best_result.posterior_winrate, 4),
            sufficiency=sufficiency,
            prior_strength=self._prior_strength(best_result.samples),
            half_life_days=self.decay.half_life,
            source="bayesian",
            avg_return_7d=best_result.avg_return,
            profit_factor=profit_factor,
            max_adverse_excursion=avg_mae,
        )

    def _global_winrate(self, conn: sqlite3.Connection,
                        signal_type: str) -> float:
        """Get global historical winrate for a signal type (Empirical Bayes prior)."""
        row = conn.execute(
            "SELECT AVG(is_win_7d) as wr, COUNT(*) as cnt "
            "FROM signal_features WHERE signal_type = ? AND is_win_7d IS NOT NULL",
            (signal_type,)
        ).fetchone()
        if row and row[1] and row[1] > 0:
            return float(row[0])
        return 0.50  # default: neutral

    def _prior_strength(self, samples: int) -> int:
        """Dynamic prior strength based on sample count."""
        if samples < 100:
            return 20  # Beta(10,10)
        elif samples < 500:
            return 40  # Beta(20,20)
        else:
            return 100  # Beta(50,50) — data dominates

    def _bayesian_posterior(self, global_rate: float, wins: float,
                            samples: int) -> float:
        """
        Empirical Bayes posterior: prior = global signal winrate.

        posterior = (alpha + wins) / (alpha + beta + samples)
        where alpha/(alpha+beta) = global_rate
        """
        prior_strength = self._prior_strength(samples)
        alpha = max(global_rate * prior_strength, 1.0)  # floor at 1 to ensure shrinkage
        beta = max((1 - global_rate) * prior_strength, 1.0)
        return (alpha + wins) / (alpha + beta + samples)

    def precompute_cache(self):
        """
        Precompute expectancy for all state_hashes in signal_features.
        Populates expectancy_cache table with raw weighted counts (not posteriors).
        """
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row

        hashes = conn.execute(
            "SELECT DISTINCT market_state_hash, signal_type FROM signal_features"
        ).fetchall()

        today = Date.today()
        count = 0

        for row in hashes:
            h = row["market_state_hash"]
            sig = row["signal_type"]

            df = pd.read_sql_query(
                "SELECT date, is_win_7d, result_7d "
                "FROM signal_features WHERE market_state_hash = ? AND signal_type = ?",
                conn, params=(h, sig)
            )

            if df.empty:
                continue

            dates_list = [Date.fromisoformat(d) for d in df["date"]]
            weights = self.decay.weights(dates_list, today)
            wins_w = (df["is_win_7d"].fillna(0).values * weights).sum()
            losses_w = ((1 - df["is_win_7d"].fillna(0)).values * weights).sum()
            ret_sum = (df["result_7d"].fillna(0).values * weights).sum()
            ret_sq = ((df["result_7d"].fillna(0).values ** 2) * weights).sum()
            eff_n = weights.sum()

            sufficiency = self.guard.evaluate(eff_n).value

            conn.execute("""
                INSERT OR REPLACE INTO expectancy_cache
                (state_hash, signal_type, wins_weighted, losses_weighted,
                 sum_return_7d, sum_return_sq_7d, effective_samples, sufficiency)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """, (h, sig, wins_w, losses_w, ret_sum, ret_sq, eff_n, sufficiency))
            count += 1

        conn.commit()
        conn.close()
        logger.info(f"Precomputed expectancy cache: {count} state×signal combos")
        return count