backtest_multi_period.py

"""Multi-Period Backtest: v1 vs PlanA vs PlanB vs PlanC
3 time periods:
  Period 1: 2024-05-01 to 2024-12-01
  Period 2: 2024-11-01 to 2025-05-01
  Period 3: 2025-05-01 to 2025-12-01
"""
import sys, logging
logging.basicConfig(level=logging.WARNING)
import numpy as np
import pandas as pd

sys.path.insert(0, 'd:/trandingagent/deer-flow/skills/custom/factor-analysis')
from tools.local_data_store import _read_csv, _build_csv_ticker_map

INITIAL_CAPITAL = 300000
COMMISSION_RATE = 0.001
STAMP_TAX_RATE = 0.001
NUM_STOCKS = 12
CANDIDATE_POOL = 24
MIN_DATA_DAYS = 120

PERIODS = [
    ("P1_2024H2", "2024-05-01", "2024-12-01"),
    ("P2_2024H2to2025H1", "2024-11-01", "2025-05-01"),
    ("P3_2025H1", "2025-05-01", "2025-12-01"),
]

GROWTH_CONFIG = {
    "Momentum_5D": {"weight": 0.20, "direction": 1},
    "Momentum_1M": {"weight": 0.25, "direction": 1},
    "Momentum_3M": {"weight": 0.20, "direction": 1},
    "volume_ratio": {"weight": 0.15, "direction": 1},
    "turnover_rate": {"weight": 0.20, "direction": 1},
}


def load_all_stocks(max_stocks=500):
    ticker_map = _build_csv_ticker_map()
    all_data = {}
    for ticker in ticker_map:
        df = _read_csv(ticker)
        if df.empty or len(df) < MIN_DATA_DAYS:
            continue
        df["date"] = pd.to_datetime(df["date"])
        df = df.set_index("date").sort_index()
        all_data[ticker] = df
    return all_data


def get_trading_fridays(all_data, start, end):
    dates = set()
    for ticker, df in all_data.items():
        mask = (df.index >= pd.Timestamp(start)) & (df.index <= pd.Timestamp(end))
        for d in df.index[mask]:
            if d.weekday() == 4:
                dates.add(d)
    return sorted(dates)


def calc_factors_at_date(df, date):
    ts = pd.Timestamp(date)
    loc = df.index.get_indexer([ts], method="ffill")[0]
    if loc < 60:
        return None
    sub = df.iloc[:loc+1].copy()
    if len(sub) < 60:
        return None

    close = sub["close"]
    vol = sub.get("volume_lots", pd.Series(dtype=float))
    turnover = sub.get("turnover_rate", pd.Series(dtype=float))

    factors = {}
    if len(close) >= 6 and close.iloc[-6] > 0:
        factors["Momentum_5D"] = (close.iloc[-1] / close.iloc[-6] - 1) * 100
    if len(close) >= 23 and close.iloc[-23] > 0:
        factors["Momentum_1M"] = (close.iloc[-1] / close.iloc[-23] - 1) * 100
    if len(close) >= 67 and close.iloc[-67] > 0:
        factors["Momentum_3M"] = (close.iloc[-1] / close.iloc[-67] - 1) * 100

    if len(close) >= 15:
        delta = close.diff()
        gain = delta.clip(lower=0).rolling(14).mean()
        loss = (-delta.clip(upper=0)).rolling(14).mean()
        rs = gain / (loss + 1e-10)
        rsi = 100 - (100 / (1 + rs))
        if pd.notna(rsi.iloc[-1]):
            factors["RSI14"] = float(rsi.iloc[-1])
        rsi_over70 = rsi > 70
        ob_duration = 0
        for j in range(len(rsi_over70) - 1, -1, -1):
            if rsi_over70.iloc[j]:
                ob_duration += 1
            else:
                break
        factors["RSI_ob_duration"] = ob_duration

    if len(close) >= 21:
        ret = close.pct_change(fill_method=None).dropna()
        if len(ret) >= 20:
            factors["Volatility"] = float(ret.iloc[-20:].std() * np.sqrt(252) * 100)

    if len(close) >= 2:
        factors["last_chg"] = float((close.iloc[-1] / close.iloc[-2] - 1) * 100)

    if turnover is not None and len(turnover) > 0 and pd.notna(turnover.iloc[-1]):
        factors["turnover_rate"] = float(turnover.iloc[-1])
    if vol is not None and len(vol) >= 6:
        avg5 = vol.iloc[-6:-1].mean()
        if avg5 > 0 and pd.notna(vol.iloc[-1]):
            factors["volume_ratio"] = float(vol.iloc[-1] / avg5)

    factors["close"] = float(close.iloc[-1])
    return {k: v for k, v in factors.items() if v is not None and pd.notna(v) and np.isfinite(v)}


def _growth_score(factor_data):
    df = pd.DataFrame(factor_data).T
    if df.empty:
        return pd.DataFrame()
    available = [c for c in GROWTH_CONFIG.keys() if c in df.columns]
    if len(available) < 2:
        return pd.DataFrame()
    valid_mask = df[available].notna().sum(axis=1) >= 2
    scoring_df = df[valid_mask].copy()
    for col in available:
        cfg = GROWTH_CONFIG[col]
        series = scoring_df[col].dropna()
        if len(series) < 5:
            continue
        q_low, q_high = series.quantile(0.01), series.quantile(0.99)
        clipped = scoring_df[col].clip(lower=q_low, upper=q_high)
        ranked = clipped.rank(pct=True)
        if cfg["direction"] == -1:
            ranked = 1 - ranked
        scoring_df[f"{col}_score"] = ranked * cfg["weight"]
    score_cols = [f"{col}_score" for col in available if f"{col}_score" in scoring_df.columns]
    if score_cols:
        scoring_df["growth_score"] = scoring_df[score_cols].sum(axis=1)
    return scoring_df


def select_v1(factor_data):
    scoring_df = _growth_score(factor_data)
    if scoring_df.empty:
        return []
    scoring_df = scoring_df.sort_values("growth_score", ascending=False)
    return [(idx, row.get("close", 0), row["growth_score"]) for idx, row in scoring_df.head(NUM_STOCKS).iterrows()]


def select_plan_a(factor_data):
    scoring_df = _growth_score(factor_data)
    if scoring_df.empty:
        return []
    scoring_df = scoring_df.sort_values("growth_score", ascending=False)
    candidates = scoring_df.head(CANDIDATE_POOL).copy()
    if "RSI_ob_duration" in candidates.columns:
        false_breakout = candidates["RSI_ob_duration"].between(1, 2)
        filtered = candidates[~false_breakout]
        if len(filtered) >= NUM_STOCKS:
            candidates = filtered
        else:
            keep_idx = filtered.index.tolist()
            removed = candidates[false_breakout].sort_values("growth_score", ascending=False)
            for idx in removed.index:
                keep_idx.append(idx)
                if len(keep_idx) >= NUM_STOCKS:
                    break
            candidates = candidates.loc[keep_idx]
    return [(idx, row.get("close", 0), row["growth_score"]) for idx, row in candidates.head(NUM_STOCKS).iterrows()]


def select_plan_b(factor_data):
    scoring_df = _growth_score(factor_data)
    if scoring_df.empty:
        return []
    scoring_df = scoring_df.sort_values("growth_score", ascending=False)
    candidates = scoring_df.head(CANDIDATE_POOL).copy()
    if "RSI_ob_duration" in candidates.columns:
        candidates["rsi_rank_score"] = 0.0
        mask_short = candidates["RSI_ob_duration"].between(1, 2)
        candidates.loc[mask_short, "rsi_rank_score"] = -0.5
        mask_normal = candidates["RSI_ob_duration"] == 0
        candidates.loc[mask_normal, "rsi_rank_score"] = 0.0
        mask_confirmed = candidates["RSI_ob_duration"] >= 3
        candidates.loc[mask_confirmed, "rsi_rank_score"] = 0.3
        mask_strong = candidates["RSI_ob_duration"] >= 6
        candidates.loc[mask_strong, "rsi_rank_score"] = 0.6
        mask_very = candidates["RSI_ob_duration"] >= 10
        candidates.loc[mask_very, "rsi_rank_score"] = 1.0
        candidates["final_score"] = candidates["growth_score"] + candidates["rsi_rank_score"]
        candidates = candidates.sort_values("final_score", ascending=False)
    return [(idx, row.get("close", 0), row.get("final_score", row["growth_score"])) for idx, row in candidates.head(NUM_STOCKS).iterrows()]


def select_plan_c(factor_data):
    scoring_df = _growth_score(factor_data)
    if scoring_df.empty:
        return [], {}
    scoring_df = scoring_df.sort_values("growth_score", ascending=False)
    top = scoring_df.head(NUM_STOCKS)
    tickers = [(idx, row.get("close", 0), row["growth_score"]) for idx, row in top.iterrows()]
    weight_map = {}
    for idx, row in top.iterrows():
        ob_dur = row.get("RSI_ob_duration", 0)
        if pd.notna(ob_dur) and 1 <= ob_dur <= 2:
            weight_map[idx] = 0.5
        else:
            weight_map[idx] = 1.0
    return tickers, weight_map


class Portfolio:
    def __init__(self, name, capital=INITIAL_CAPITAL):
        self.name = name
        self.initial_capital = capital
        self.cash = capital
        self.positions = {}
        self.history = []
        self.trade_log = []
        self.peak_value = capital
        self.max_drawdown = 0

    def get_price(self, all_data, ticker, date):
        df = all_data.get(ticker)
        if df is None:
            return None
        ts = pd.Timestamp(date)
        try:
            if ts in df.index:
                return float(df.loc[ts, "close"])
            loc = df.index.get_indexer([ts], method="ffill")[0]
            if loc >= 0:
                return float(df.iloc[loc]["close"])
        except:
            pass
        return None

    def total_value(self, all_data, date):
        total = self.cash
        for ticker, pos in self.positions.items():
            price = self.get_price(all_data, ticker, date)
            if price and np.isfinite(price):
                total += pos["shares"] * price
        return total

    def rebalance(self, all_data, date, target_tickers, date_label="", weight_map=None):
        current_value = self.total_value(all_data, date)
        if not np.isfinite(current_value) or current_value <= 0:
            return

        if weight_map:
            total_weight = sum(weight_map.get(t, 1.0) for t, _, _ in target_tickers)
            alloc = {}
            for ticker, price, score in target_tickers:
                w = weight_map.get(ticker, 1.0)
                alloc[ticker] = current_value * (w / total_weight)
        else:
            target_value_per_stock = current_value / max(len(target_tickers), 1)
            alloc = {t: target_value_per_stock for t, _, _ in target_tickers}

        target_set = set(t for t, _, _ in target_tickers)
        for ticker in list(self.positions.keys()):
            if ticker not in target_set:
                price = self.get_price(all_data, ticker, date)
                if price and price > 0:
                    shares = self.positions[ticker]["shares"]
                    value = shares * price
                    cost = value * COMMISSION_RATE + value * STAMP_TAX_RATE
                    self.cash += value - cost
                    self.trade_log.append({"date": date_label, "ticker": ticker, "action": "SELL", "shares": shares, "price": round(price, 2)})
                del self.positions[ticker]
            else:
                price = self.get_price(all_data, ticker, date)
                if price and price > 0:
                    current_pos_value = self.positions[ticker]["shares"] * price
                    target_val = alloc.get(ticker, current_value / max(len(target_tickers), 1))
                    diff_value = target_val - current_pos_value
                    if abs(diff_value) > current_value * 0.05:
                        if diff_value > 0:
                            buy_shares = int(diff_value / price / 100) * 100
                            if buy_shares > 0:
                                cost_val = buy_shares * price
                                commission = cost_val * COMMISSION_RATE
                                total_cost = cost_val + commission
                                if total_cost <= self.cash:
                                    self.cash -= total_cost
                                    self.positions[ticker]["shares"] += buy_shares
                                    self.trade_log.append({"date": date_label, "ticker": ticker, "action": "BUY", "shares": buy_shares, "price": round(price, 2)})
                        else:
                            sell_shares = int(abs(diff_value) / price / 100) * 100
                            if sell_shares > 0 and sell_shares <= self.positions[ticker]["shares"]:
                                sell_value = sell_shares * price
                                cost = sell_value * COMMISSION_RATE + sell_value * STAMP_TAX_RATE
                                self.cash += sell_value - cost
                                self.positions[ticker]["shares"] -= sell_shares
                                self.trade_log.append({"date": date_label, "ticker": ticker, "action": "SELL", "shares": sell_shares, "price": round(price, 2)})

        for ticker, price, score in target_tickers:
            if ticker in self.positions:
                continue
            if not price or price <= 0 or not np.isfinite(price):
                continue
            target_val = alloc.get(ticker, current_value / max(len(target_tickers), 1))
            shares = int(target_val / price / 100) * 100
            if shares <= 0:
                shares = 100
            cost_val = shares * price
            commission = cost_val * COMMISSION_RATE
            total_cost = cost_val + commission
            if total_cost > self.cash:
                shares = int(self.cash / (price * (1 + COMMISSION_RATE)) / 100) * 100
                if shares <= 0:
                    continue
                cost_val = shares * price
                commission = cost_val * COMMISSION_RATE
                total_cost = cost_val + commission
            self.cash -= total_cost
            self.positions[ticker] = {"shares": shares, "buy_price": price}
            self.trade_log.append({"date": date_label, "ticker": ticker, "action": "BUY", "shares": shares, "price": round(price, 2)})

    def record(self, all_data, date, date_label=""):
        value = self.total_value(all_data, date)
        if not np.isfinite(value) or value <= 0:
            value = self.cash
        if value > self.peak_value:
            self.peak_value = value
        dd = (value - self.peak_value) / self.peak_value * 100
        if dd < self.max_drawdown:
            self.max_drawdown = dd
        self.history.append({
            "date": date_label, "total_value": round(value, 2), "cash": round(self.cash, 2),
            "return_pct": round((value / self.initial_capital - 1) * 100, 2),
            "drawdown_pct": round(dd, 2), "num_positions": len(self.positions),
        })

    def summary(self):
        if not self.history:
            return {}
        final = self.history[-1]
        weekly_returns = []
        for i in range(1, len(self.history)):
            prev_val = self.history[i-1]["total_value"]
            curr_val = self.history[i]["total_value"]
            if prev_val > 0:
                weekly_returns.append(curr_val / prev_val - 1)
        if weekly_returns:
            ann_ret = (1 + np.mean(weekly_returns)) ** 52 - 1
            ann_vol = np.std(weekly_returns) * np.sqrt(52)
            sharpe = ann_ret / ann_vol if ann_vol > 0 else 0
        else:
            ann_ret = ann_vol = sharpe = 0
        total_trades = len(self.trade_log)
        buy_trades = len([t for t in self.trade_log if t["action"] == "BUY"])
        return {
            "strategy": self.name, "final_value": final["total_value"],
            "total_return_pct": final["return_pct"],
            "max_return_pct": round(max(h["return_pct"] for h in self.history), 2),
            "max_drawdown_pct": round(self.max_drawdown, 2),
            "annualized_return": round(ann_ret * 100, 2),
            "annualized_volatility": round(ann_vol * 100, 2),
            "sharpe_ratio": round(sharpe, 2),
            "total_trades": total_trades, "buy_trades": buy_trades,
        }


def run_period(all_data, period_name, start_date, end_date):
    fridays = get_trading_fridays(all_data, start_date, end_date)
    if not fridays:
        print(f"\n  [{period_name}] No trading fridays found, skipping")
        return {}

    print(f"\n  [{period_name}] {start_date} to {end_date}, {len(fridays)} weeks")

    strategies = {
        "v1_纯成长": Portfolio(f"{period_name}_v1", INITIAL_CAPITAL),
        "A_过滤假突破": Portfolio(f"{period_name}_A", INITIAL_CAPITAL),
        "B_两阶段排序": Portfolio(f"{period_name}_B", INITIAL_CAPITAL),
        "C_仓位调节": Portfolio(f"{period_name}_C", INITIAL_CAPITAL),
    }

    for i, friday in enumerate(fridays):
        date_str = friday.strftime("%Y-%m-%d")
        week_num = i + 1

        factor_data = {}
        for ticker in all_data:
            try:
                f = calc_factors_at_date(all_data[ticker], friday)
                if f and len(f) >= 3:
                    factor_data[ticker] = f
            except:
                pass

        if not factor_data:
            for p in strategies.values():
                p.record(all_data, friday, date_str)
            continue

        v1_tickers = select_v1(factor_data)
        strategies["v1_纯成长"].rebalance(all_data, friday, v1_tickers, date_str)

        plan_a_tickers = select_plan_a(factor_data)
        strategies["A_过滤假突破"].rebalance(all_data, friday, plan_a_tickers, date_str)

        plan_b_tickers = select_plan_b(factor_data)
        strategies["B_两阶段排序"].rebalance(all_data, friday, plan_b_tickers, date_str)

        plan_c_tickers, weight_map = select_plan_c(factor_data)
        strategies["C_仓位调节"].rebalance(all_data, friday, plan_c_tickers, date_str, weight_map=weight_map)

        for p in strategies.values():
            p.record(all_data, friday, date_str)

        if week_num % 8 == 0:
            h = strategies["v1_纯成长"].history[-1]
            print(f"    Week {week_num}/{len(fridays)}: v1={h['return_pct']:+.2f}%")

    results = {}
    for name, p in strategies.items():
        s = p.summary()
        s["period"] = period_name
        s["strategy"] = name
        results[name] = s

    print(f"\n  [{period_name}] Results:")
    for name, s in results.items():
        print(f"    {name}: Return={s['total_return_pct']:+.2f}% MDD={s['max_drawdown_pct']:.2f}% Sharpe={s['sharpe_ratio']:.2f}")

    return results


def main():
    print("Loading stock data...")
    all_data = load_all_stocks()
    print(f"  Loaded {len(all_data)} stocks")

    all_results = []
    for period_name, start, end in PERIODS:
        results = run_period(all_data, period_name, start, end)
        if results:
            all_results.extend(results.values())

    print(f"\n{'='*90}")
    print(f"  MULTI-PERIOD COMPARISON SUMMARY")
    print(f"{'='*90}")

    if not all_results:
        print("  No results to display")
        return

    df = pd.DataFrame(all_results)

    metrics = ["total_return_pct", "max_drawdown_pct", "sharpe_ratio", "total_trades"]
    strategies = ["v1_纯成长", "A_过滤假突破", "B_两阶段排序", "C_仓位调节"]

    for period_name, _, _ in PERIODS:
        period_df = df[df["period"] == period_name]
        if period_df.empty:
            continue
        print(f"\n  [{period_name}]")
        header = f"  {'Strategy':<16}" + "".join(f" {m:<20}" for m in metrics)
        print(header)
        print(f"  {'-'*16}" + "".join(f" {'-'*20}" for _ in metrics))
        for strat in strategies:
            row = period_df[period_df["strategy"] == strat]
            if row.empty:
                continue
            row = row.iloc[0]
            line = f"  {strat:<16}"
            for m in metrics:
                line += f" {row[m]:<20.2f}"
            print(line)

    print(f"\n{'='*90}")
    print(f"  AGGREGATE RANKING (across all periods)")
    print(f"{'='*90}")

    sharpe_by_strategy = {}
    return_by_strategy = {}
    mdd_by_strategy = {}
    for strat in strategies:
        strat_df = df[df["strategy"] == strat]
        if strat_df.empty:
            continue
        sharpes = strat_df["sharpe_ratio"].tolist()
        returns = strat_df["total_return_pct"].tolist()
        mdds = strat_df["max_drawdown_pct"].tolist()
        sharpe_by_strategy[strat] = np.mean(sharpes)
        return_by_strategy[strat] = np.mean(returns)
        mdd_by_strategy[strat] = np.mean(mdds)

    ranked = sorted(sharpe_by_strategy.items(), key=lambda x: x[1], reverse=True)
    print(f"\n  {'Rank':<6} {'Strategy':<16} {'Avg Sharpe':<12} {'Avg Return':<12} {'Avg MDD':<12}")
    print(f"  {'-'*6} {'-'*16} {'-'*12} {'-'*12} {'-'*12}")
    for rank, (strat, avg_sharpe) in enumerate(ranked, 1):
        avg_ret = return_by_strategy[strat]
        avg_mdd = mdd_by_strategy[strat]
        marker = " <-- BEST" if rank == 1 else ""
        print(f"  #{rank:<5} {strat:<16} {avg_sharpe:<12.2f} {avg_ret:<+12.2f} {avg_mdd:<12.2f}{marker}")

    win_counts = {s: 0 for s in strategies}
    for period_name, _, _ in PERIODS:
        period_df = df[df["period"] == period_name]
        if period_df.empty:
            continue
        best = period_df.loc[period_df["sharpe_ratio"].idxmax(), "strategy"]
        win_counts[best] = win_counts.get(best, 0) + 1

    print(f"\n  Period wins (best Sharpe in each period):")
    for strat in strategies:
        print(f"    {strat}: {win_counts.get(strat, 0)}/{len(PERIODS)} periods")

    df.to_csv("d:/trandingagent/backtest_multi_period_results.csv", index=False, encoding="utf-8-sig")
    print(f"\n  Results saved: d:/trandingagent/backtest_multi_period_results.csv")


if __name__ == "__main__":
    main()