services/backtest_service.py

"""
Walk-forward backtest service.

Extracted from routers/stock.py (#21) to separate business logic from the router layer.
"""

import logging
from datetime import datetime

from dateutil.relativedelta import relativedelta

from data.fetcher import fetch_history, fetch_cross_asset_tw, is_us_ticker
from data.institutional_flow import add_institutional_flow
from indicators.technical import add_all_indicators, add_cross_asset_tw
from models.predictor import (
    _build_features,
    FEATURE_COLUMNS,
    StockPredictor,
    _adaptive_thresholds,
    _as_lgbm_feature_frame,
)

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Trading cost constants (Taiwan stock market)
# ---------------------------------------------------------------------------
COMMISSION_PCT = 0.1425       # broker commission per side
COMMISSION_DISCOUNT = 0.6     # typical online broker discount
TAX_PCT = 0.3                 # securities transaction tax (sell-side only)
BUY_COST = COMMISSION_PCT * COMMISSION_DISCOUNT        # 0.0855%
SELL_COST = COMMISSION_PCT * COMMISSION_DISCOUNT + TAX_PCT  # 0.3855%
ROUND_TRIP_COST_PCT = BUY_COST + SELL_COST             # 0.471%

MIN_TRAIN = 120    # minimum training window (trading days)
HORIZON = 5        # prediction horizon (trading days)
RETRAIN_EVERY = 5  # retrain every N prediction days
EMBARGO = HORIZON  # skip HORIZON days before prediction point to prevent look-ahead
TRAIN_WINDOW = 250 # sliding training window (1 trading year); use all data if shorter


class BacktestError(Exception):
    """Raised when backtest cannot proceed (e.g. no data)."""
    pass


def compute_net_return(signal: str, actual_return: float, cost: float) -> float:
    """
    Compute net return for a given signal.

    - BUY  → long position: actual_return - cost
    - SELL → short position: -actual_return - cost
    - HOLD → no trade: 0
    """
    if signal == "BUY":
        return actual_return - cost
    elif signal == "SELL":
        return -actual_return - cost
    else:
        return 0.0


def walk_forward_backtest(stock_no: str, months: int) -> dict:
    """
    Walk-forward backtest with embargo period and sliding training window.

    Key improvements:
    - Embargo: train on data[:i-EMBARGO] instead of data[:i], eliminating the
      look-ahead bias where the last training target overlaps with the test point.
    - Dynamic window: use a sliding TRAIN_WINDOW (250 days) instead of the full
      history, so the model adapts to recent market regimes.
    - Cross-asset features: TAIEX + USD/TWD merged before feature extraction.

    Raises BacktestError if no data is found.
    """
    # Fetch extra history for training warmup
    df = fetch_history(stock_no, months=months + 12)
    if df.empty:
        raise BacktestError(f"No data found for {stock_no}")

    df = add_all_indicators(df)

    # Cross-asset enrichment (Taiwan stocks only)
    if not is_us_ticker(stock_no):
        df = add_institutional_flow(df, stock_no)
        start = str(df["date"].min())
        end   = str(df["date"].max())
        taiex, usdtwd, sox, tnx = fetch_cross_asset_tw(start, end)
        df = add_cross_asset_tw(df, taiex, usdtwd, sox_close=sox, tnx_close=tnx)

    feat = _build_features(df)

    cutoff = str((datetime.today() - relativedelta(months=months)).date())

    results = []
    predictor = None
    last_train_idx = -RETRAIN_EVERY  # force first train

    for i in range(MIN_TRAIN + EMBARGO, len(df) - HORIZON):
        row_feat = feat.iloc[i]
        if row_feat[FEATURE_COLUMNS].isna().any():
            continue

        date_str = str(df["date"].iloc[i])
        if date_str < cutoff:
            continue

        # Embargo: exclude the HORIZON rows immediately before prediction point
        # so no training target leaks into the test window.
        train_end = i - EMBARGO
        train_start = max(0, train_end - TRAIN_WINDOW)

        if train_end - train_start < MIN_TRAIN:
            continue

        # Retrain if needed (every RETRAIN_EVERY prediction days)
        if i - last_train_idx >= RETRAIN_EVERY or predictor is None:
            try:
                predictor = StockPredictor()
                predictor.train(
                    df.iloc[train_start:train_end],
                    precomputed_features=feat.iloc[train_start:train_end],
                )
                last_train_idx = i
            except Exception:
                continue

        current_close = float(df["close"].iloc[i])
        future_close = float(df["close"].iloc[i + HORIZON])
        actual_return = (future_close - current_close) / current_close * 100

        X = row_feat[FEATURE_COLUMNS].values.reshape(1, -1)
        X_scaled = predictor.scaler.transform(X)
        X_scaled_lgbm = _as_lgbm_feature_frame(X_scaled)
        def _class_one_probability(model, X_input=X_scaled) -> float:
            prob = model.predict_proba(X_input)[0]
            return float(prob[1]) if len(prob) > 1 else 0.5

        buy_probs = [_class_one_probability(predictor.classifier)]
        sell_probs = [_class_one_probability(predictor.sell_classifier)]

        if getattr(predictor, "_use_xgb", False) and predictor.xgb_clf is not None:
            try:
                buy_probs.append(_class_one_probability(predictor.xgb_clf))
                sell_probs.append(_class_one_probability(predictor.xgb_sell))
            except Exception:
                pass
        if getattr(predictor, "_use_lgbm", False) and predictor.lgbm_clf is not None:
            try:
                buy_probs.append(_class_one_probability(predictor.lgbm_clf, X_scaled_lgbm))
                sell_probs.append(_class_one_probability(predictor.lgbm_sell, X_scaled_lgbm))
            except Exception:
                pass

        buy_prob = sum(buy_probs) / len(buy_probs)
        sell_prob = sum(sell_probs) / len(sell_probs)
        vol_20d = float(row_feat.get("volatility_20d", 0) or 0)
        buy_threshold, sell_threshold = _adaptive_thresholds(vol_20d)

        if buy_prob >= buy_threshold:
            signal = "BUY"
        elif sell_prob >= (1 - sell_threshold):
            signal = "SELL"
        else:
            signal = "HOLD"

        correct = (signal == "BUY" and actual_return > 0) or \
                  (signal == "SELL" and actual_return < 0)

        net_return = compute_net_return(signal, actual_return, ROUND_TRIP_COST_PCT)

        results.append({
            "date": date_str,
            "signal": signal,
            "buy_prob": round(buy_prob, 3),
            "sell_prob": round(sell_prob, 3),
            "close": round(current_close, 2),
            "future_close": round(future_close, 2),
            "actual_return_pct": round(actual_return, 2),
            "net_return_pct": round(net_return, 2),
            "correct": correct,
        })

    return _build_stats(stock_no, months, results)


def _build_stats(stock_no: str, months: int, results: list[dict]) -> dict:
    """Aggregate backtest results into summary statistics."""
    buys = [r for r in results if r["signal"] == "BUY"]
    sells = [r for r in results if r["signal"] == "SELL"]
    holds = [r for r in results if r["signal"] == "HOLD"]
    non_holds = [r for r in results if r["signal"] != "HOLD"]

    def win_rate(lst):
        return round(sum(1 for r in lst if r["correct"]) / len(lst) * 100, 1) if lst else 0

    def avg_ret(lst):
        return round(sum(r["net_return_pct"] for r in lst) / len(lst), 2) if lst else 0

    def max_consecutive(lst, correct_val: bool):
        max_streak = cur_streak = 0
        for r in lst:
            if r["correct"] == correct_val:
                cur_streak += 1
                max_streak = max(max_streak, cur_streak)
            else:
                cur_streak = 0
        return max_streak

    trade_net_returns = [r["net_return_pct"] for r in non_holds]

    if non_holds:
        signal_accuracy_overall = round(
            sum(1 for r in non_holds if r["correct"]) / len(non_holds) * 100, 1
        )
    else:
        signal_accuracy_overall = 0

    return {
        "stock_no": stock_no,
        "months": months,
        "walk_forward": True,
        "trading_cost_pct": round(ROUND_TRIP_COST_PCT, 4),
        "cost_breakdown": {
            "buy_commission": round(BUY_COST, 4),
            "sell_commission_and_tax": round(SELL_COST, 4),
        },
        "stats": {
            "total_days": len(results),
            "buy_signals": len(buys),
            "sell_signals": len(sells),
            "hold_signals": len(holds),
            "buy_win_rate": win_rate(buys),
            "sell_win_rate": win_rate(sells),
            "avg_return_on_buy": avg_ret(buys),
            "avg_return_on_sell": avg_ret(sells),
            "total_return_if_followed": round(sum(r["net_return_pct"] for r in non_holds), 2),
            "gross_return": round(sum(r["actual_return_pct"] for r in buys), 2),
            "max_consecutive_wins": max_consecutive(non_holds, True),
            "max_consecutive_losses": max_consecutive(non_holds, False),
            "best_trade_pct": round(max(trade_net_returns), 2) if trade_net_returns else 0,
            "worst_trade_pct": round(min(trade_net_returns), 2) if trade_net_returns else 0,
            "signal_accuracy_overall": signal_accuracy_overall,
        },
        "trades": results,
    }