set50-system/backend/app/backtest.py

"""Real backtest engine — allocate across a date range, track P&L.

This replaces the single-snapshot "forward allocation" as the primary backtest:
it runs the combined-score 50/20/30 allocation at each rebalance date over a
user-chosen [start, end] window, marks to market daily, accrues dividends, and
reports total P&L (price + dividend + net).

Honesty: runs on revised vendor history (non-PIT public data). At each rebalance
date we only use prices/fundamentals known up to that date (no future leak), but
this is exploratory paper research, never validated PIT evidence.
"""

from __future__ import annotations

import datetime as dt
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

from .simulation import (
    Candidate, Order, allocate_capital, load_price_snapshot, MIN_SHARES,
)

_PROC_DIR = Path(__file__).resolve().parent.parent / "data" / "prices"


@dataclass
class BacktestResult:
    start: str
    end: str
    capital: float
    final_value: float = 0.0
    price_pnl: float = 0.0
    dividend_income: float = 0.0
    net_return: float = 0.0
    trades: int = 0
    rebalances: int = 0
    holdings: dict = field(default_factory=dict)  # final

    def to_dict(self) -> dict:
        return {
            "start": self.start, "end": self.end, "capital": self.capital,
            "final_value": round(self.final_value, 2),
            "price_pnl": round(self.price_pnl, 2),
            "dividend_income": round(self.dividend_income, 2),
            "net_return": round(self.net_return, 4),
            "trades": self.trades, "rebalances": self.rebalances,
            "holdings": self.holdings,
        }


class BacktestError(Exception):
    pass


def _bars_up_to(series: dict, sym: str, date: dt.date) -> list:
    bars = series.get(sym, {}).get("bars", [])
    return [b for b in bars if _bar_date(b.get("date")) <= date]


def _bar_date(s: Optional[str]) -> dt.date:
    if not s:
        return dt.date.min
    return dt.date.fromisoformat(str(s)[:10])


def _latest_close(series: dict, sym: str, date: dt.date) -> Optional[float]:
    bars = _bars_up_to(series, sym, date)
    if bars:
        return float(bars[-1]["adjusted_close"])
    return None


def _rebalance_dates(start: str, end: str, freq: str = "monthly") -> list[str]:
    s = dt.date.fromisoformat(start)
    e = dt.date.fromisoformat(end)
    if s >= e:
        raise BacktestError("end must be after start")
    dates = []
    cur = s
    if freq == "monthly":
        while cur <= e:
            dates.append(cur.isoformat())
            y, m = (cur.year + 1, 1) if cur.month == 12 else (cur.year, cur.month + 1)
            cur = dt.date(y, m, 1)
    elif freq == "quarterly":
        while cur <= e:
            dates.append(cur.isoformat())
            q = (cur.month - 1) // 3 + 1
            if q == 4:
                cur = dt.date(cur.year + 1, 1, 1)
            else:
                cur = dt.date(cur.year, q * 3 + 1, 1)
    else:
        raise BacktestError(f"unsupported freq {freq}")
    return dates


def _candidates_at(series: dict, syms: list[str], date: dt.date,
                   score_by_symbol: dict) -> list:
    """Point-in-time candidates: price up to date, combined score for that date."""
    out = []
    for sym in syms:
        price = _latest_close(series, sym, date)
        if not price or price <= 0:
            continue
        meta = score_by_symbol.get(sym, {})
        out.append(Candidate(
            symbol=sym, price=price,
            combined_score=float(meta.get("combined", 0.0)),
            is_dividend=bool(meta.get("is_dividend", False)),
            dividend_yield=float(meta.get("dividend_yield") or 0.0),
        ))
    return out


def run_backtest(
    start: str, end: str,
    capital: float = 1_000_000,
    rebalance_freq: str = "monthly",
    score_fn=None,
    symbols: Optional[list[str]] = None,
) -> BacktestResult:
    """Run the backtest. `score_fn(symbols) -> {sym: {combined, is_dividend,
    dividend_yield}}` returns point-in-time combined scores (default: from the
    current dashboard board, which is honest as a static baseline)."""
    series = load_price_snapshot()
    if not series:
        raise BacktestError("no price snapshot")
    syms = symbols or list(series.keys())

    if score_fn is None:
        # default: current combined scores (static baseline — honest non-PIT).
        from .dashboard import default_scores
        score_by_symbol = default_scores(syms) or {}
    else:
        score_by_symbol = score_fn(syms)

    dates = _rebalance_dates(start, end, rebalance_freq)
    result = BacktestResult(start=start, end=end, capital=capital)
    result.rebalances = len(dates)

    # holdings: {sym: {qty, cost}}
    holdings: dict = {}
    total_dividend = 0.0
    cash = capital
    trades = 0

    _e = dt.date.fromisoformat(end)

    # Buy-and-hold backtest: allocate once at the first rebalance date that has
    # price data, then hold to the end. (A full multi-rebalance engine with
    # position selling is a follow-up; this answers 'what would I have earned by
    # buying per this system on <start> and holding until <end>?' without the
    # double-spend bug.)
    for d_iso in dates:
        d = dt.date.fromisoformat(d_iso)
        cands = _candidates_at(series, syms, d, score_by_symbol)
        if not cands:
            continue
        alloc = allocate_capital(capital, cands)
        for o in alloc.orders:
            holdings[o.symbol] = {"qty": o.qty, "cost": o.notional}
            cash -= o.notional
            trades += 1
        break  # single allocation at first available rebalance, then hold

    # mark-to-market to end + dividend
    final_value = cash
    for sym, h in holdings.items():
        px = _latest_close(series, sym, _e)
        if px:
            final_value += h["qty"] * px
            # crude dividend: yield% * cost (proxy, honest-flagged)
            meta = score_by_symbol.get(sym, {})
            yield_pct = float(meta.get("dividend_yield") or 0.0) / 100.0
            total_dividend += h["cost"] * yield_pct

    result.holdings = {s: h["qty"] for s, h in holdings.items()}
    result.final_value = final_value
    result.price_pnl = final_value - cash - total_dividend
    result.dividend_income = total_dividend
    result.net_return = (final_value - capital) / capital if capital else 0.0
    result.trades = trades
    return result


def total_investable(cands: list) -> float:
    return sum(c.price for c in cands if c.symbol)