trumpsignal-backend/app/services/backtest.py

"""
Single-post backtest harness.

Given a Post with a directional signal (buy/short) and a target asset, fetch
the historical 1-minute candles for [published_at, published_at + max_hold_h]
from Binance and replay the new convex-strategy exit rules. Outputs what
WOULD have happened if we had been live at that moment with the current
trailing-stop / stop-loss / max-hold settings.

Why this exists:
  The bot's m1h accuracy is 45.7%, but that was measured with a 1-hour
  fixed-window snapshot — i.e. assuming we close at exactly the 1-hour mark.
  The new exit logic (trailing stop, 7-day max hold) is FUNDAMENTALLY
  different. Without backtesting we can't know whether changing exits also
  changes the win rate / PnL profile. This harness lets us check.

Scope (deliberately narrow for the MVP):
  - One post at a time. Batch runner sits on top.
  - Uses 1m HIGH/LOW within each bar (worst-case path) — conservative.
  - Ignores fees + slippage (caller is expected to subtract them).
  - Assumes immediate fill at the candle's OPEN on entry.
  - Does NOT re-evaluate regime gates (the caller decides whether to
    include the post). This makes "what would have happened" cleaner.

The 1m granularity matters: with 5m or 1H bars you can't distinguish
"stop loss hit then bounced back" from "rallied straight up", and a
trailing-stop strategy lives or dies on that distinction.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta, timezone
from typing import Optional

import httpx

logger = logging.getLogger(__name__)


BINANCE_KLINES_URL = "https://api.binance.com/api/v3/klines"


@dataclass
class BacktestResult:
    post_id:        int
    asset:          str
    side:           str           # "long" | "short"
    entry_price:    float
    exit_price:     float
    exit_reason:    str           # "trailing_stop" | "stop_loss" | "take_profit" | "max_hold"
    hold_minutes:   int
    pnl_pct:        float         # net of nothing — apply your own fee model
    peak_pct:       float         # best unrealised gain reached
    trough_pct:     float         # worst unrealised gain reached
    bars_evaluated: int

    def to_dict(self) -> dict:
        return asdict(self)


@dataclass
class BacktestParams:
    """Exit-rule snapshot used for the simulation."""
    stop_loss_pct:            float = 1.5
    trailing_stop_pct:        Optional[float] = 2.5
    trailing_activate_at_pct: Optional[float] = 5.0
    take_profit_pct:          Optional[float] = None
    max_hold_hours:           int   = 168


async def fetch_binance_1m(
    symbol: str,
    start: datetime,
    end: datetime,
    client: Optional[httpx.AsyncClient] = None,
) -> list[dict]:
    """Pull 1-minute candles from Binance spot for [start, end].

    Binance caps a single call at 1000 candles → we page in chunks. Returns
    candles in chronological order. Each candle: {time_ms, open, high, low, close}.
    """
    own_client = client is None
    if own_client:
        client = httpx.AsyncClient(timeout=20)
    out: list[dict] = []
    cursor_ms = int(start.replace(tzinfo=timezone.utc).timestamp() * 1000)
    end_ms    = int(end.replace(tzinfo=timezone.utc).timestamp() * 1000)
    try:
        while cursor_ms < end_ms:
            params = {
                "symbol":   symbol,
                "interval": "1m",
                "startTime": cursor_ms,
                "endTime":   end_ms,
                "limit":    1000,
            }
            resp = await client.get(BINANCE_KLINES_URL, params=params)
            resp.raise_for_status()
            chunk = resp.json()
            if not chunk:
                break
            for row in chunk:
                out.append({
                    "time_ms": row[0],
                    "open":    float(row[1]),
                    "high":    float(row[2]),
                    "low":     float(row[3]),
                    "close":   float(row[4]),
                })
            # Next page starts after the last candle returned.
            last_open = chunk[-1][0]
            if last_open <= cursor_ms:
                break
            cursor_ms = last_open + 60_000
            if len(chunk) < 1000:
                break  # final partial page
    finally:
        if own_client:
            await client.aclose()
    return out


def _asset_to_symbol(asset: str) -> str:
    """Map our internal asset code to a Binance spot symbol.

    Most assets are simply {ASSET}USDT. TRUMP/HYPE/etc. may not exist on
    Binance — those will fail at fetch time and we'll return None upstream.
    """
    return f"{asset.upper()}USDT"


def simulate_exit(
    candles: list[dict],
    side: str,
    params: BacktestParams,
) -> dict:
    """Walk the candles 1m at a time, applying the exit ladder.

    Priority within a bar (worst-case ordering):
      1. Stop loss  — assume hit via LOW (long) / HIGH (short)
      2. Trailing   — only if armed; assume hit via the same worst-case extreme
      3. Take profit — fixed TP if set
    A real bar can't tell us whether the high or low printed first, so we
    take the pessimistic path: STOP LOSS BEFORE PROFIT if both could have
    triggered. This makes the backtest conservative.
    """
    if not candles:
        return {"reason": "no_data", "exit_price": 0.0, "bars": 0,
                "peak_pct": 0.0, "trough_pct": 0.0, "pnl_pct": 0.0}

    entry = candles[0]["open"]
    if entry == 0:
        return {"reason": "bad_entry", "exit_price": 0.0, "bars": 0,
                "peak_pct": 0.0, "trough_pct": 0.0, "pnl_pct": 0.0}

    is_long = side == "long"
    peak_pct   = 0.0
    trough_pct = 0.0
    armed      = False

    def pct(price: float) -> float:
        raw = (price - entry) / entry
        return (raw if is_long else -raw) * 100

    for i, bar in enumerate(candles):
        # Within-bar extremes in position's favoured direction
        good_extreme = bar["high"] if is_long else bar["low"]
        bad_extreme  = bar["low"]  if is_long else bar["high"]
        good_pct = pct(good_extreme)
        bad_pct  = pct(bad_extreme)

        # Update running peak / trough using extremes
        if good_pct > peak_pct:   peak_pct   = good_pct
        if bad_pct  < trough_pct: trough_pct = bad_pct

        # 1. Stop loss — pessimistic check first
        if bad_pct <= -params.stop_loss_pct:
            # Approximate exit at the SL trigger price (not the bar extreme)
            sl_price = entry * (1 - params.stop_loss_pct / 100) if is_long \
                       else entry * (1 + params.stop_loss_pct / 100)
            return {
                "reason": "stop_loss",
                "exit_price": sl_price,
                "bars": i + 1,
                "peak_pct": peak_pct,
                "trough_pct": trough_pct,
                "pnl_pct": -params.stop_loss_pct,
            }

        # 2. Trailing — arm if peak crossed activation
        if (params.trailing_stop_pct is not None
            and params.trailing_activate_at_pct is not None):
            if not armed and peak_pct >= params.trailing_activate_at_pct:
                armed = True
            if armed:
                # Drawdown from peak (using bad_extreme of this bar)
                drawdown = peak_pct - bad_pct
                if drawdown >= params.trailing_stop_pct:
                    # Exit at peak − trailing_stop_pct (approx)
                    exit_pct = peak_pct - params.trailing_stop_pct
                    ts_price = entry * (1 + exit_pct / 100) if is_long \
                               else entry * (1 - exit_pct / 100)
                    return {
                        "reason": "trailing_stop",
                        "exit_price": ts_price,
                        "bars": i + 1,
                        "peak_pct": peak_pct,
                        "trough_pct": trough_pct,
                        "pnl_pct": exit_pct,
                    }

        # 3. Fixed TP
        if params.take_profit_pct is not None and good_pct >= params.take_profit_pct:
            tp_price = entry * (1 + params.take_profit_pct / 100) if is_long \
                       else entry * (1 - params.take_profit_pct / 100)
            return {
                "reason": "take_profit",
                "exit_price": tp_price,
                "bars": i + 1,
                "peak_pct": peak_pct,
                "trough_pct": trough_pct,
                "pnl_pct": params.take_profit_pct,
            }

    # Walked the whole window without hitting any rule → close at last bar
    last_close = candles[-1]["close"]
    return {
        "reason": "max_hold",
        "exit_price": last_close,
        "bars": len(candles),
        "peak_pct": peak_pct,
        "trough_pct": trough_pct,
        "pnl_pct": pct(last_close),
    }


async def backtest_post(
    post_id: int,
    params: Optional[BacktestParams] = None,
) -> Optional[BacktestResult]:
    """Backtest a single Post by ID.

    Returns None if:
      - Post not found
      - signal is not buy/short
      - target_asset is unknown / not on Binance
      - Binance has no data for the window
    """
    from app.database import AsyncSessionLocal
    from app.models import Post
    from sqlalchemy import select

    params = params or BacktestParams()

    async with AsyncSessionLocal() as db:
        result = await db.execute(select(Post).where(Post.id == post_id))
        post = result.scalar_one_or_none()

    if post is None:
        logger.warning("Backtest: post %d not found", post_id)
        return None
    if post.signal not in ("buy", "short"):
        logger.warning("Backtest: post %d signal=%s not actionable", post_id, post.signal)
        return None

    asset = post.target_asset or post.price_impact_asset
    if not asset:
        logger.warning("Backtest: post %d has no asset", post_id)
        return None

    side = "long" if post.signal == "buy" else "short"
    symbol = _asset_to_symbol(asset)

    # Window: [published_at, published_at + max_hold_hours]
    start = post.published_at
    end   = start + timedelta(hours=params.max_hold_hours)
    # Don't backtest a window that hasn't fully elapsed yet
    now_naive = datetime.now(timezone.utc).replace(tzinfo=None)
    if end > now_naive:
        end = now_naive

    candles = await fetch_binance_1m(symbol, start, end)
    if not candles:
        logger.warning("Backtest: no Binance data for %s in [%s, %s]", symbol, start, end)
        return None

    sim = simulate_exit(candles, side, params)
    if sim["reason"] in ("no_data", "bad_entry"):
        return None

    return BacktestResult(
        post_id=post_id,
        asset=asset,
        side=side,
        entry_price=candles[0]["open"],
        exit_price=sim["exit_price"],
        exit_reason=sim["reason"],
        hold_minutes=sim["bars"],
        pnl_pct=round(sim["pnl_pct"], 3),
        peak_pct=round(sim["peak_pct"], 3),
        trough_pct=round(sim["trough_pct"], 3),
        bars_evaluated=sim["bars"],
    )


async def backtest_batch(
    limit: int = 50,
    min_confidence: int = 80,
    params: Optional[BacktestParams] = None,
) -> dict:
    """Backtest every directional post with confidence ≥ min_confidence.

    Returns aggregate stats + per-trade results. Posts whose Binance data
    is missing (TRUMP, niche perps) are skipped silently.
    """
    from app.database import AsyncSessionLocal
    from app.models import Post
    from sqlalchemy import select, and_, or_

    params = params or BacktestParams()

    async with AsyncSessionLocal() as db:
        rows = await db.execute(
            select(Post.id).where(
                and_(
                    Post.signal.in_(["buy", "short"]),
                    Post.ai_confidence >= min_confidence,
                )
            ).order_by(Post.published_at.desc()).limit(limit)
        )
        post_ids = [r[0] for r in rows.all()]

    results: list[BacktestResult] = []
    skipped: list[int] = []
    for pid in post_ids:
        try:
            r = await backtest_post(pid, params)
            if r is None:
                skipped.append(pid)
            else:
                results.append(r)
        except Exception as exc:
            logger.error("Backtest post %d failed: %s", pid, exc)
            skipped.append(pid)

    if not results:
        return {"params": asdict(params), "results": [], "skipped": skipped,
                "summary": {"trades": 0}}

    pnls = [r.pnl_pct for r in results]
    wins = [p for p in pnls if p > 0]
    losses = [p for p in pnls if p <= 0]
    by_reason: dict[str, int] = {}
    for r in results:
        by_reason[r.exit_reason] = by_reason.get(r.exit_reason, 0) + 1

    summary = {
        "trades":       len(results),
        "skipped":      len(skipped),
        "win_rate_pct": round(len(wins) / len(results) * 100, 1),
        "avg_pnl_pct":  round(sum(pnls) / len(pnls), 3),
        "total_pnl_pct": round(sum(pnls), 3),
        "best_pct":     round(max(pnls), 3),
        "worst_pct":    round(min(pnls), 3),
        "avg_win_pct":  round(sum(wins) / len(wins), 3) if wins else 0.0,
        "avg_loss_pct": round(sum(losses) / len(losses), 3) if losses else 0.0,
        "exit_reasons": by_reason,
    }
    return {
        "params":  asdict(params),
        "summary": summary,
        "skipped": skipped,
        "results": [r.to_dict() for r in results],
    }