Finding Statistical Edge: How AI Agents Identify Exploitable Inefficiencies

import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import List, Dict, Callable, Optional, Tuple
from enum import Enum

class Side(Enum):
    LONG = "long"
    SHORT = "short"

class FillMode(Enum):
    OPTIMISTIC = "optimistic"   # mid-price fill (overfit prone)
    REALISTIC  = "realistic"    # cross spread + slippage
    PESSIMISTIC = "pessimistic"  # worst-case for stress testing

@dataclass
class Trade:
    entry_time: pd.Timestamp
    exit_time:  pd.Timestamp
    symbol:     str
    side:       Side
    entry_price: float
    exit_price:  float
    size:        float
    fee_entry:   float
    fee_exit:    float
    slippage:    float

    @property
    def pnl(self) -> float:
        direction = 1 if self.side == Side.LONG else -1
        gross = direction * (self.exit_price - self.entry_price) * self.size
        costs = (self.fee_entry + self.fee_exit + self.slippage) * self.size
        return gross - costs

    @property
    def pnl_pct(self) -> float:
        return self.pnl / (self.entry_price * self.size)


@dataclass
class BacktestConfig:
    initial_capital:   float = 10_000.0
    commission_rate:   float = 0.001    # 10 bps per side
    slippage_model:    str   = "sqrt"   # sqrt of dollar volume
    slippage_factor:   float = 0.1      # market impact coefficient
    max_position_pct:  float = 0.10     # max 10% of capital per trade
    fill_mode:         FillMode = FillMode.REALISTIC
    risk_free_rate:    float = 0.045    # 4.5% annualized


class BacktestEngine:
    """
    Honest backtesting engine that models realistic execution costs.
    Designed to produce conservative estimates of live performance.
    """

    def __init__(self, config: BacktestConfig):
        self.config = config
        self.trades: List[Trade] = []
        self.equity_curve: List[float] = []
        self.capital = config.initial_capital

    def _calc_slippage(self, price: float, size: float, adv: float) -> float:
        """
        Market impact via square-root model.
        slippage = factor * sigma * sqrt(size / ADV) * price
        """
        sigma = 0.02  # assumed daily vol; refine with realized vol
        participation = size / max(adv, 1e-9)
        impact = self.config.slippage_factor * sigma * np.sqrt(participation) * price
        if self.config.fill_mode == FillMode.PESSIMISTIC:
            impact *= 2.0
        elif self.config.fill_mode == FillMode.OPTIMISTIC:
            impact *= 0.0
        return impact

    def run(
        self,
        prices: pd.DataFrame,    # columns: open, high, low, close, volume
        signals: pd.Series,      # +1 long, -1 short, 0 flat per bar
        symbol: str = "ASSET",
    ) -> "BacktestResult":
        position = 0
        entry_price = 0.0
        entry_time = None
        entry_size = 0.0
        self.equity_curve = [self.capital]

        for i, (ts, row) in enumerate(prices.iterrows()):
            sig = signals.iloc[i] if i < len(signals) else 0
            close = row['close']
            adv = row.get('volume', 1e6) * close  # dollar volume

            # Exit existing position
            if position != 0 and (sig != position or i == len(prices) - 1):
                slip = self._calc_slippage(close, entry_size, adv)
                fee = close * entry_size * self.config.commission_rate
                side = Side.LONG if position > 0 else Side.SHORT
                t = Trade(
                    entry_time=entry_time, exit_time=ts,
                    symbol=symbol, side=side,
                    entry_price=entry_price, exit_price=close,
                    size=entry_size,
                    fee_entry=entry_price * entry_size * self.config.commission_rate,
                    fee_exit=fee, slippage=slip,
                )
                self.capital += t.pnl
                self.trades.append(t)
                position = 0

            # Enter new position
            if sig != 0 and position == 0:
                max_notional = self.capital * self.config.max_position_pct
                entry_size = max_notional / close
                slip = self._calc_slippage(close, entry_size, adv)
                entry_price = close + (slip if sig > 0 else -slip)
                entry_time = ts
                position = int(sig)

            self.equity_curve.append(self.capital)

        return BacktestResult(self.trades, self.equity_curve, self.config)


class BacktestResult:
    def __init__(self, trades: List[Trade], equity: List[float], cfg: BacktestConfig):
        self.trades = trades
        self.equity = np.array(equity)
        self.cfg = cfg

    def sharpe(self) -> float:
        returns = np.diff(self.equity) / self.equity[:-1]
        excess = returns - self.cfg.risk_free_rate / 252
        return np.sqrt(252) * excess.mean() / (excess.std() + 1e-9)

    def max_drawdown(self) -> float:
        peak = np.maximum.accumulate(self.equity)
        dd = (self.equity - peak) / peak
        return dd.min()

    def expectancy(self) -> float:
        pnls = [t.pnl for t in self.trades]
        if not pnls: return 0.0
        wins = [p for p in pnls if p > 0]
        losses = [p for p in pnls if p <= 0]
        p_win = len(wins) / len(pnls)
        avg_win  = np.mean(wins)   if wins   else 0.0
        avg_loss = abs(np.mean(losses)) if losses else 0.0
        return p_win * avg_win - (1 - p_win) * avg_loss

    def summary(self) -> Dict:
        return {
            "total_trades":   len(self.trades),
            "sharpe":          round(self.sharpe(), 3),
            "max_drawdown":    round(self.max_drawdown() * 100, 2),
            "expectancy_usd":  round(self.expectancy(), 4),
            "total_return_pct": round((self.equity[-1] / self.equity[0] - 1) * 100, 2),
            "win_rate":         round(
                len([t for t in self.trades if t.pnl > 0]) / max(len(self.trades), 1), 3
            ),
        }

import numpy as np
from scipy import stats
from typing import List, Dict, Any

class OverfitDetector:
    """
    Statistical tests to detect overfitting before capital deployment.
    Run these on every strategy before going live.
    """

    def __init__(self, n_permutations: int = 1000):
        self.n_permutations = n_permutations

    def deflated_sharpe(
        self,
        observed_sharpe: float,
        n_trials: int,
        n_observations: int,
        skewness: float = 0.0,
        kurtosis: float = 3.0,
    ) -> float:
        """
        Deflated Sharpe Ratio (Bailey & Lopez de Prado, 2014).
        Returns probability that the strategy has a true positive Sharpe.
        """
        # Expected maximum Sharpe from n_trials independent tests
        eulers_const = 0.5772156649
        expected_max = (
            (1 - eulers_const) * stats.norm.ppf(1 - 1/n_trials)
            + eulers_const * stats.norm.ppf(1 - 1/(n_trials * np.e))
        )

        # Variance of Sharpe ratio (non-normal corrections)
        sr_var = (1 / n_observations) * (
            1
            - skewness * observed_sharpe
            + ((kurtosis - 1) / 4) * observed_sharpe**2
        )

        # Probability that true SR > 0 after deflation
        z = (observed_sharpe - expected_max) / np.sqrt(sr_var)
        prob = stats.norm.cdf(z)
        return float(prob)

    def permutation_test(
        self,
        strategy_returns: np.ndarray,
        metric_fn: callable,
    ) -> Dict[str, float]:
        """
        Null hypothesis: the strategy has no edge.
        Under H0, shuffling returns should produce equivalent results.
        """
        observed = metric_fn(strategy_returns)
        null_distribution = []

        for _ in range(self.n_permutations):
            shuffled = np.random.permutation(strategy_returns)
            null_distribution.append(metric_fn(shuffled))

        null_arr = np.array(null_distribution)
        p_value = np.mean(null_arr >= observed)

        return {
            "observed":  float(observed),
            "p_value":   float(p_value),
            "null_mean": float(null_arr.mean()),
            "null_std":  float(null_arr.std()),
            "z_score":   float((observed - null_arr.mean()) / (null_arr.std() + 1e-9)),
            "significant": bool(p_value < 0.05),
        }

    def parameter_stability(
        self,
        param_range: np.ndarray,
        sharpe_scores: np.ndarray,
        threshold: float = 0.5,
    ) -> Dict[str, Any]:
        """
        A genuine edge should work across a range of parameter values.
        If only a narrow window produces positive results, it's overfit.
        """
        positive_pct = np.mean(sharpe_scores > threshold)
        peak_sharpe = sharpe_scores.max()
        peak_param = param_range[np.argmax(sharpe_scores)]

        # Gradient smoothness: overfitting creates sharp peaks
        gradient = np.abs(np.gradient(sharpe_scores))
        smoothness = 1.0 / (gradient.mean() + 1e-9)

        return {
            "positive_pct": float(positive_pct),
            "peak_sharpe":  float(peak_sharpe),
            "peak_param":   float(peak_param),
            "smoothness":   float(smoothness),
            "robust": bool(positive_pct > 0.6 and smoothness > 1.0),
        }

    def full_audit(
        self,
        is_returns:  np.ndarray,   # in-sample returns
        oos_returns: np.ndarray,   # out-of-sample returns
        n_trials:    int = 50,
    ) -> Dict[str, Any]:
        """Run the complete overfitting audit."""
        def sharpe_fn(r): return np.sqrt(252) * r.mean() / (r.std() + 1e-9)

        is_sharpe  = sharpe_fn(is_returns)
        oos_sharpe = sharpe_fn(oos_returns)
        dsr = self.deflated_sharpe(is_sharpe, n_trials, len(is_returns))
        perm = self.permutation_test(is_returns, sharpe_fn)

        verdict = (
            "PASS" if (
                oos_sharpe > 0.7 * is_sharpe
                and dsr > 0.95
                and perm["significant"]
            ) else "FAIL"
        )

        return {
            "verdict":    verdict,
            "is_sharpe":  round(is_sharpe, 3),
            "oos_sharpe": round(oos_sharpe, 3),
            "is_oos_ratio": round(oos_sharpe / (is_sharpe + 1e-9), 3),
            "deflated_sr_prob": round(dsr, 4),
            "permutation_p": round(perm["p_value"], 4),
        }

import numpy as np
import pandas as pd
from typing import Callable, Dict, List, Tuple, Any
from itertools import product
from concurrent.futures import ThreadPoolExecutor
import warnings

class WalkForwardOptimizer:
    """
    Anchored and rolling walk-forward optimization.
    This is the correct way to validate parameter-dependent strategies.
    """

    def __init__(
        self,
        train_periods:  int,        # bars in training window
        test_periods:   int,        # bars in test window
        anchored:       bool = False, # rolling vs anchored
        n_workers:      int = 4,
    ):
        self.train = train_periods
        self.test  = test_periods
        self.anchored = anchored
        self.n_workers = n_workers

    def _windows(self, n: int) -> List[Tuple[int, int, int, int]]:
        """Generate (train_start, train_end, test_start, test_end) indices."""
        windows = []
        test_start = self.train
        while test_start + self.test <= n:
            train_start = 0 if self.anchored else test_start - self.train
            windows.append((train_start, test_start, test_start, test_start + self.test))
            test_start += self.test
        return windows

    def optimize(
        self,
        data:       pd.DataFrame,
        strategy_fn: Callable[[pd.DataFrame, Dict], pd.Series],
        param_grid: Dict[str, List[Any]],
        objective:  Callable[[pd.Series], float],
    ) -> "WFOResult":
        """
        Run full WFO across all windows.
        strategy_fn(data, params) -> signal series
        objective(returns) -> scalar to maximize
        """
        windows = self._windows(len(data))
        all_param_combos = [
            dict(zip(param_grid.keys(), vals))
            for vals in product(*param_grid.values())
        ]

        oos_returns_list = []
        best_params_per_window = []

        for win_idx, (tr_s, tr_e, ts_s, ts_e) in enumerate(windows):
            train_data = data.iloc[tr_s:tr_e]
            test_data  = data.iloc[ts_s:ts_e]

            # Optimize on training window (parallel)
            def score_params(params):
                try:
                    sig = strategy_fn(train_data, params)
                    returns = train_data['close'].pct_change() * sig.shift(1)
                    return objective(returns.dropna())
                except Exception:
                    return -np.inf

            with ThreadPoolExecutor(max_workers=self.n_workers) as ex:
                scores = list(ex.map(score_params, all_param_combos))

            best_idx = int(np.argmax(scores))
            best_params = all_param_combos[best_idx]
            best_params_per_window.append({
                "window": win_idx,
                "params": best_params,
                "is_score": scores[best_idx],
            })

            # Apply best params to unseen test window
            oos_sig = strategy_fn(test_data, best_params)
            oos_ret = test_data['close'].pct_change() * oos_sig.shift(1)
            oos_returns_list.append(oos_ret.dropna())

        oos_combined = pd.concat(oos_returns_list)
        return WFOResult(oos_combined, best_params_per_window)


class WFOResult:
    def __init__(self, oos_returns: pd.Series, param_history: List[Dict]):
        self.returns = oos_returns
        self.param_history = param_history

    def sharpe(self) -> float:
        r = self.returns
        return float(np.sqrt(252) * r.mean() / (r.std() + 1e-9))

    def parameter_stability_score(self) -> float:
        """
        How stable are the optimal parameters across windows?
        Higher = more stable = less overfitting.
        """
        if not self.param_history: return 0.0
        all_params = [h["params"] for h in self.param_history]
        scores = []
        for key in all_params[0]:
            vals = [p[key] for p in all_params if isinstance(p[key], (int, float))]
            if vals:
                cv = np.std(vals) / (np.mean(vals) + 1e-9)
                scores.append(1.0 / (1.0 + cv))
        return float(np.mean(scores)) if scores else 0.0

    def summary(self) -> Dict:
        eq = (1 + self.returns).cumprod()
        peak = eq.cummax()
        dd = ((eq - peak) / peak).min()
        return {
            "oos_sharpe":          round(self.sharpe(), 3),
            "oos_total_return_pct": round((eq.iloc[-1] - 1) * 100, 2),
            "oos_max_drawdown_pct": round(dd * 100, 2),
            "param_stability":     round(self.parameter_stability_score(), 3),
            "n_windows":           len(self.param_history),
        }

import numpy as np
from collections import deque
from dataclasses import dataclass, field
from typing import Optional
import asyncio, aiohttp, time

@dataclass
class EdgeState:
    rolling_pnl:     deque = field(default_factory=lambda: deque(maxlen=100))
    rolling_win_rate: deque = field(default_factory=lambda: deque(maxlen=100))
    degraded:        bool = False
    halt_triggered:  bool = False
    last_alert_ts:   float = 0.0

class LiveEdgeMonitor:
    """
    Real-time edge health monitoring for deployed agents.
    Connects to Purple Flea trading API for live P&L stream.
    """
    BASE = "https://purpleflea.com/trading-api"

    def __init__(
        self,
        api_key: str,
        expected_sharpe:      float,    # backtest Sharpe
        degradation_threshold: float = 0.5, # halt if below this pct of expected
        min_trades:           int = 30,    # min trades before checks activate
    ):
        self.api_key = api_key
        self.expected_sharpe = expected_sharpe
        self.threshold = degradation_threshold
        self.min_trades = min_trades
        self.state = EdgeState()

    async def record_trade(self, pnl: float, win: bool) -> Optional[str]:
        """Record a completed trade and check edge health. Returns alert if degraded."""
        self.state.rolling_pnl.append(pnl)
        self.state.rolling_win_rate.append(float(win))

        if len(self.state.rolling_pnl) < self.min_trades:
            return None

        live_sharpe = self._rolling_sharpe()
        win_rate = np.mean(self.state.rolling_win_rate)
        degradation_ratio = live_sharpe / (self.expected_sharpe + 1e-9)

        # Alert conditions
        if degradation_ratio < self.threshold:
            alert = (
                f"EDGE ALERT: Live Sharpe {live_sharpe:.2f} vs expected "
                f"{self.expected_sharpe:.2f} "
                f"({degradation_ratio*100:.0f}% of backtest). "
                f"Win rate: {win_rate*100:.1f}%. CONSIDER HALTING."
            )
            self.state.degraded = True
            if degradation_ratio < self.threshold * 0.5:
                self.state.halt_triggered = True
                alert = "HALT: " + alert
            return alert

        self.state.degraded = False
        return None

    def _rolling_sharpe(self) -> float:
        r = np.array(self.state.rolling_pnl)
        return float(np.sqrt(252) * r.mean() / (r.std() + 1e-9))

    async def fetch_recent_trades(self, session: aiohttp.ClientSession, limit: int = 50):
        """Pull recent trades from Purple Flea trading API."""
        async with session.get(
            f"{self.BASE}/trades",
            params={"limit": limit},
            headers={"Authorization": f"Bearer {self.api_key}"},
        ) as resp:
            if resp.status == 200:
                data = await resp.json()
                return data.get("trades", [])
            return []

    def status(self) -> dict:
        n = len(self.state.rolling_pnl)
        return {
            "n_trades_tracked": n,
            "live_sharpe": round(self._rolling_sharpe(), 3) if n >= self.min_trades else None,
            "win_rate": round(float(np.mean(self.state.rolling_win_rate)), 3) if n > 0 else None,
            "degraded": self.state.degraded,
            "halt_triggered": self.state.halt_triggered,
        }