wehub-resource-sync
664 lines
28 KiB
Python
664 lines
28 KiB
Python
"""LRS TQQQ Strategy — Historical Backtest Engine.
|
||
|
||
Compares LRS strategy vs QQQ buy-and-hold across six time windows:
|
||
近 2 年 / 近 3 年 / 近 5 年 / 近 10 年 / 2015—2020 / 2010—2015
|
||
|
||
Key design choices
|
||
──────────────────
|
||
• Each window downloads 1 extra year for MA200 warmup before trading starts.
|
||
• "Initial entry": if signal is already True on day 1 of the trading window,
|
||
enter immediately (no need to wait for a fresh MA200 crossover).
|
||
• Transaction cost: $1.99 / side (Futu US stock rates).
|
||
• Initial capital: $10,000 USD.
|
||
• Idle cash earns 4% annualized, compounded each trading day (≈ (1.04)^(1/252) per day).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timedelta
|
||
|
||
import matplotlib.pyplot as plt
|
||
import numpy as np
|
||
import pandas as pd
|
||
import streamlit as st
|
||
import yfinance as yf
|
||
|
||
from web.pages.lrs.strategy_shared import apply_dark_style
|
||
|
||
_INITIAL_CAPITAL = 10_000.0
|
||
_COST_PER_SIDE = 1.99 # $0.99 commission + $1.00 platform fee (Futu)
|
||
_DISTANCE_THRESH = 6.0 # macro_score=3 → 6% (Step 3)
|
||
_WHIPSAW_DAYS = 10 # Holding ≤ N days AND loss
|
||
_CASH_YIELD_ANNUAL = 0.04 # idle cash MMF / T-bill style; compounded on trading days
|
||
_CASH_YIELD_DAILY = (1.0 + _CASH_YIELD_ANNUAL) ** (1.0 / 252.0)
|
||
|
||
|
||
def _accrue_cash_yield(cash: float) -> float:
|
||
"""Apply one trading day of compound interest on idle cash (4% annualized)."""
|
||
return cash * _CASH_YIELD_DAILY
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Period configuration
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
#
|
||
# Each entry:
|
||
# label – displayed in the sub-tab
|
||
# fetch_type – "period" (yfinance period string) or "range" (start/end dates)
|
||
# dl_period – yfinance period to download (includes warmup) [fetch_type=period]
|
||
# dl_start – warmup start date string [fetch_type=range]
|
||
# dl_end – end date string [fetch_type=range]
|
||
# target_days – approx trading days to keep after warmup [used when eff_start=None]
|
||
# eff_start – fixed effective start date (None = compute from target_days / TQQQ)
|
||
|
||
_PERIOD_SPECS: dict[str, dict] = {
|
||
"2y": {
|
||
"label": "近 2 年", "fetch_type": "period",
|
||
"dl_period": "3y", "target_days": 504,
|
||
},
|
||
"3y": {
|
||
"label": "近 3 年", "fetch_type": "period",
|
||
"dl_period": "4y", "target_days": 756,
|
||
},
|
||
"5y": {
|
||
"label": "近 5 年", "fetch_type": "period",
|
||
"dl_period": "6y", "target_days": 1260,
|
||
},
|
||
"10y": {
|
||
"label": "近 10 年", "fetch_type": "range",
|
||
"dl_start": None, "dl_end": None, # computed at runtime (today − 11 yr)
|
||
"target_days": 2520,
|
||
},
|
||
"2015_2020": {
|
||
"label": "2015—2020", "fetch_type": "range",
|
||
"dl_start": "2014-01-01", "dl_end": "2020-12-31",
|
||
"eff_start": "2015-01-02",
|
||
},
|
||
"2010_2015": {
|
||
"label": "2010—2015", "fetch_type": "range",
|
||
"dl_start": "2009-01-01", "dl_end": "2015-12-31",
|
||
"eff_start": None, # will use first available TQQQ date (~2010-02-09)
|
||
},
|
||
}
|
||
|
||
_PERIOD_ORDER = ["2y", "3y", "5y", "10y", "2015_2020", "2010_2015"]
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Data fetching
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
@st.cache_data(ttl=3600, show_spinner=False)
|
||
def _fetch_by_period(dl_period: str) -> tuple[pd.DataFrame, pd.DataFrame]:
|
||
"""Download by yfinance period string (includes warmup year)."""
|
||
qqq = yf.download("QQQ", period=dl_period, auto_adjust=True, progress=False)
|
||
tqqq = yf.download("TQQQ", period=dl_period, auto_adjust=True, progress=False)
|
||
for df in (qqq, tqqq):
|
||
if isinstance(df.columns, pd.MultiIndex):
|
||
df.columns = df.columns.get_level_values(0)
|
||
return qqq, tqqq
|
||
|
||
|
||
@st.cache_data(ttl=3600, show_spinner=False)
|
||
def _fetch_by_range(start: str, end: str) -> tuple[pd.DataFrame, pd.DataFrame]:
|
||
"""Download by explicit date range (includes warmup year)."""
|
||
qqq = yf.download("QQQ", start=start, end=end, auto_adjust=True, progress=False)
|
||
tqqq = yf.download("TQQQ", start=start, end=end, auto_adjust=True, progress=False)
|
||
for df in (qqq, tqqq):
|
||
if isinstance(df.columns, pd.MultiIndex):
|
||
df.columns = df.columns.get_level_values(0)
|
||
return qqq, tqqq
|
||
|
||
|
||
def _fetch(period_key: str) -> tuple[pd.DataFrame, pd.DataFrame]:
|
||
"""Dispatch to the right fetch function."""
|
||
spec = _PERIOD_SPECS[period_key]
|
||
if spec["fetch_type"] == "period":
|
||
return _fetch_by_period(spec["dl_period"])
|
||
# date-range based
|
||
start = spec.get("dl_start")
|
||
end = spec.get("dl_end")
|
||
if start is None: # "10y": compute dynamically
|
||
end_dt = datetime.now()
|
||
start_dt = end_dt - timedelta(days=365 * 11)
|
||
start = start_dt.strftime("%Y-%m-%d")
|
||
end = end_dt.strftime("%Y-%m-%d")
|
||
return _fetch_by_range(start, end)
|
||
|
||
|
||
def _get_effective_start(
|
||
qqq_full: pd.Series,
|
||
tqqq_full: pd.Series,
|
||
period_key: str,
|
||
) -> pd.Timestamp:
|
||
"""Return the first tradeable date (after warmup, after TQQQ inception)."""
|
||
spec = _PERIOD_SPECS[period_key]
|
||
|
||
# Fixed start defined in config
|
||
if spec.get("eff_start"):
|
||
ts = pd.Timestamp(spec["eff_start"])
|
||
# Clamp to actual data range
|
||
return max(ts, qqq_full.index[0])
|
||
|
||
# "2010_2015": TQQQ only starts ~Feb 2010 — use first TQQQ date
|
||
if period_key == "2010_2015":
|
||
return tqqq_full.index[0]
|
||
|
||
# Period-based / "10y": keep last target_days trading days
|
||
n = spec.get("target_days", 504)
|
||
offset = max(0, len(qqq_full) - n)
|
||
return qqq_full.index[offset]
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Signals
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _streak(s: pd.Series) -> pd.Series:
|
||
"""Vectorized consecutive-True count (resets to 0 on False)."""
|
||
groups = (s != s.shift()).cumsum()
|
||
return (s.groupby(groups).cumcount() + 1).where(s, 0)
|
||
|
||
|
||
def _build_signals(qqq_close: pd.Series) -> pd.DataFrame:
|
||
ma200 = qqq_close.rolling(200).mean()
|
||
a1 = qqq_close > ma200
|
||
slope = (ma200 - ma200.shift(20)) / ma200.shift(20) * 100
|
||
a2 = slope > -0.5 # MA200 flat or rising
|
||
a3 = _streak(a1) >= 3 # 3+ consecutive closes above
|
||
|
||
signal = a1 & a2 & a3
|
||
entry_signal = signal & ~signal.shift(1).fillna(False) # False→True transition
|
||
|
||
below = qqq_close < ma200
|
||
exit_signal = below & below.shift(1).fillna(False) # 2 consecutive days below
|
||
|
||
cross = a1 & ~a1.shift(1).fillna(False)
|
||
|
||
return pd.DataFrame({
|
||
"qqq": qqq_close, "ma200": ma200,
|
||
"a1": a1, "a2": a2, "a3": a3,
|
||
"signal": signal,
|
||
"entry_signal": entry_signal,
|
||
"exit_signal": exit_signal,
|
||
"cross": cross,
|
||
})
|
||
|
||
|
||
def _allocation(dist: float) -> tuple[str, float]:
|
||
"""(mode_label, cash_fraction) per Step 3 with macro_score=3."""
|
||
if dist <= _DISTANCE_THRESH:
|
||
return "Direct Buy", 2 / 3
|
||
elif dist <= 15.0:
|
||
return "Split", 1 / 3
|
||
else:
|
||
return "Split(>15%)", 1 / 3
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Trade
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
@dataclass
|
||
class Trade:
|
||
entry_date: pd.Timestamp
|
||
entry_price: float
|
||
entry_mode: str
|
||
shares: float
|
||
cost_buy: float
|
||
exit_date: pd.Timestamp | None = None
|
||
exit_price: float | None = None
|
||
cost_sell: float = 0.0
|
||
forced_exit: bool = False
|
||
|
||
@property
|
||
def holding_days(self) -> int:
|
||
if self.exit_date is None:
|
||
return 0
|
||
return (self.exit_date - self.entry_date).days
|
||
|
||
@property
|
||
def gross_pnl(self) -> float:
|
||
if self.exit_price is None:
|
||
return 0.0
|
||
return (self.exit_price - self.entry_price) * self.shares
|
||
|
||
@property
|
||
def net_pnl(self) -> float:
|
||
return self.gross_pnl - self.cost_buy - self.cost_sell
|
||
|
||
@property
|
||
def return_pct(self) -> float:
|
||
invested = self.entry_price * self.shares + self.cost_buy
|
||
return (self.net_pnl / invested * 100) if invested > 0 else 0.0
|
||
|
||
@property
|
||
def is_whipsaw(self) -> bool:
|
||
return self.holding_days <= _WHIPSAW_DAYS and self.net_pnl < 0
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Simulation
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _simulate(
|
||
sig: pd.DataFrame,
|
||
tqqq_close: pd.Series,
|
||
trade_start: pd.Timestamp | None = None,
|
||
) -> tuple[list[Trade], pd.Series]:
|
||
"""
|
||
Simulate LRS trades day-by-day. Returns (trades, equity_series).
|
||
|
||
trade_start: first tradeable day (warmup period is skipped).
|
||
Initial-entry: if signal is already True on trade_start, enter immediately.
|
||
"""
|
||
tqqq = tqqq_close.reindex(sig.index, method="ffill")
|
||
cross_dates = sig.index[sig["cross"]].tolist()
|
||
|
||
cash = _INITIAL_CAPITAL
|
||
shares = 0.0
|
||
in_pos = False
|
||
current : Trade | None = None
|
||
trades : list[Trade] = []
|
||
equity : dict = {}
|
||
|
||
first_day = True
|
||
|
||
for date in sig.index:
|
||
if trade_start is not None and date < trade_start:
|
||
continue
|
||
|
||
cash = _accrue_cash_yield(cash)
|
||
|
||
raw = tqqq.get(date)
|
||
if raw is None or (isinstance(raw, float) and math.isnan(raw)):
|
||
equity[date] = cash + shares * (
|
||
float(tqqq.dropna().iloc[-1]) if shares > 0 else 0.0
|
||
)
|
||
first_day = False
|
||
continue
|
||
price = float(raw)
|
||
|
||
# ── Exit ──────────────────────────────────────────────────────────
|
||
if in_pos and current is not None and bool(sig.at[date, "exit_signal"]):
|
||
cash += shares * price - _COST_PER_SIDE
|
||
current.exit_date = date
|
||
current.exit_price = price
|
||
current.cost_sell = _COST_PER_SIDE
|
||
trades.append(current)
|
||
shares = 0.0
|
||
in_pos = False
|
||
current = None
|
||
|
||
# ── Entry ─────────────────────────────────────────────────────────
|
||
# Fire on normal False→True transition, or on day 1 if already bullish
|
||
is_entry = bool(sig.at[date, "entry_signal"])
|
||
if first_day and not in_pos and bool(sig.at[date, "signal"]):
|
||
is_entry = True
|
||
first_day = False
|
||
|
||
if not in_pos and is_entry:
|
||
prior = [d for d in cross_dates if d <= date]
|
||
cross_date = prior[-1] if prior else date
|
||
cross_raw = tqqq.get(cross_date)
|
||
cross_price = (
|
||
float(cross_raw)
|
||
if cross_raw is not None and not math.isnan(float(cross_raw))
|
||
else price
|
||
)
|
||
dist = (price - cross_price) / cross_price * 100 if cross_price > 0 else 0.0
|
||
mode, frac = _allocation(dist)
|
||
|
||
invest = cash * frac - _COST_PER_SIDE
|
||
if invest > 0:
|
||
n = math.floor(invest / price)
|
||
if n >= 1:
|
||
cash -= n * price + _COST_PER_SIDE
|
||
shares = float(n)
|
||
in_pos = True
|
||
current = Trade(
|
||
entry_date=date, entry_price=price,
|
||
entry_mode=mode, shares=shares, cost_buy=_COST_PER_SIDE,
|
||
)
|
||
|
||
equity[date] = cash + shares * price
|
||
|
||
# ── Force-close at period end ─────────────────────────────────────────
|
||
if in_pos and current is not None and shares > 0:
|
||
last_date = sig.index[-1]
|
||
last_price = float(tqqq.iloc[-1])
|
||
cash += shares * last_price - _COST_PER_SIDE
|
||
current.exit_date = last_date
|
||
current.exit_price = last_price
|
||
current.cost_sell = _COST_PER_SIDE
|
||
current.forced_exit = True
|
||
trades.append(current)
|
||
equity[last_date] = cash
|
||
|
||
return trades, pd.Series(equity)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Metrics
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _max_dd(equity: pd.Series) -> float:
|
||
peak = equity.cummax()
|
||
return float(((equity - peak) / peak).min() * 100)
|
||
|
||
|
||
def _metrics(trades: list[Trade], equity: pd.Series, qqq_close: pd.Series) -> dict:
|
||
n = len(trades)
|
||
wins = [t for t in trades if t.net_pnl > 0]
|
||
loss = [t for t in trades if t.net_pnl <= 0]
|
||
wsaw = [t for t in trades if t.is_whipsaw]
|
||
avg = float(np.mean([t.holding_days for t in trades])) if trades else 0.0
|
||
return {
|
||
"strategy_return": (equity.iloc[-1] / _INITIAL_CAPITAL - 1) * 100,
|
||
"qqq_return": (qqq_close.iloc[-1] / qqq_close.iloc[0] - 1) * 100,
|
||
"final_value": float(equity.iloc[-1]),
|
||
"n_trades": n,
|
||
"n_wins": len(wins),
|
||
"n_losses": len(loss),
|
||
"win_rate": (len(wins) / n * 100) if n > 0 else 0.0,
|
||
"max_dd": _max_dd(equity),
|
||
"avg_days": avg,
|
||
"n_whipsaws": len(wsaw),
|
||
"total_cost": sum(t.cost_buy + t.cost_sell for t in trades),
|
||
}
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Charts (all-English labels to avoid mojibake)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _fig_equity(
|
||
equity: pd.Series,
|
||
qqq_close: pd.Series,
|
||
trades: list[Trade],
|
||
period_label: str,
|
||
) -> plt.Figure:
|
||
qqq_bh = qqq_close / qqq_close.iloc[0] * _INITIAL_CAPITAL
|
||
idx = qqq_close.index
|
||
eq = equity.reindex(idx, method="ffill").bfill()
|
||
bh = qqq_bh.reindex(idx, method="ffill")
|
||
|
||
fig, ax = plt.subplots(figsize=(12, 5))
|
||
ax.plot(idx, eq.values, color="#38bdf8", linewidth=1.8, label="LRS Strategy")
|
||
ax.plot(idx, bh.values, color="#a78bfa", linewidth=1.4,
|
||
linestyle="--", label="QQQ Buy & Hold")
|
||
|
||
ax.fill_between(idx, eq.values, bh.values,
|
||
where=(eq.values >= bh.values), alpha=0.10,
|
||
color="#22c55e", interpolate=True)
|
||
ax.fill_between(idx, eq.values, bh.values,
|
||
where=(eq.values < bh.values), alpha=0.10,
|
||
color="#ef4444", interpolate=True)
|
||
|
||
for t in trades:
|
||
ax.axvline(t.entry_date, color="#22c55e", alpha=0.20, linewidth=0.6, linestyle=":")
|
||
if t.exit_date:
|
||
c = "#ef4444" if t.net_pnl < 0 else "#22c55e"
|
||
ax.axvline(t.exit_date, color=c, alpha=0.15, linewidth=0.6, linestyle=":")
|
||
|
||
ax.set_title(
|
||
f"Equity Curve — LRS vs QQQ Buy & Hold | {period_label}"
|
||
f" (Initial ${_INITIAL_CAPITAL:,.0f})"
|
||
)
|
||
ax.set_ylabel("Portfolio Value (USD)")
|
||
ax.set_xlabel("Date")
|
||
ax.yaxis.set_major_formatter(plt.FuncFormatter(lambda x, _: f"${x:,.0f}"))
|
||
ax.legend(fontsize=9)
|
||
apply_dark_style(fig, ax)
|
||
plt.tight_layout()
|
||
return fig
|
||
|
||
|
||
def _fig_per_trade(trades: list[Trade], period_label: str) -> plt.Figure:
|
||
if not trades:
|
||
fig, ax = plt.subplots(figsize=(8, 3))
|
||
ax.text(0.5, 0.5, f"No trades in this period ({period_label})",
|
||
ha="center", va="center", transform=ax.transAxes,
|
||
color="#94a3b8", fontsize=12)
|
||
apply_dark_style(fig, ax)
|
||
return fig
|
||
|
||
labels = [t.entry_date.strftime("%y-%m-%d") for t in trades]
|
||
returns = [t.return_pct for t in trades]
|
||
colors = ["#f59e0b" if t.is_whipsaw else ("#22c55e" if r >= 0 else "#ef4444")
|
||
for t, r in zip(trades, returns)]
|
||
|
||
fig, ax = plt.subplots(figsize=(max(10, len(trades) * 1.3), 4))
|
||
ax.bar(range(len(trades)), returns, color=colors, width=0.7, edgecolor="none")
|
||
ax.axhline(0, color="#64748b", linewidth=0.8)
|
||
ax.set_xticks(range(len(trades)))
|
||
ax.set_xticklabels(labels, rotation=45, ha="right", fontsize=8)
|
||
ax.set_ylabel("Return (%)")
|
||
ax.set_title(
|
||
f"Per-Trade Return (%) | {period_label}"
|
||
" (green=win / red=loss / orange=whipsaw)"
|
||
)
|
||
for i, (r, t) in enumerate(zip(returns, trades)):
|
||
va = "bottom" if r >= 0 else "top"
|
||
off = 0.3 if r >= 0 else -0.3
|
||
ax.text(i, r + off, f"{r:.1f}%", ha="center", va=va,
|
||
fontsize=7, color="#e2e8f0")
|
||
apply_dark_style(fig, ax)
|
||
plt.tight_layout()
|
||
return fig
|
||
|
||
|
||
def _fig_drawdown(equity: pd.Series, period_label: str) -> plt.Figure:
|
||
peak = equity.cummax()
|
||
dd = (equity - peak) / peak * 100
|
||
|
||
fig, ax = plt.subplots(figsize=(12, 3))
|
||
ax.fill_between(dd.index, dd.values, 0, color="#ef4444", alpha=0.4)
|
||
ax.plot(dd.index, dd.values, color="#f87171", linewidth=1.0)
|
||
ax.axhline(0, color="#64748b", linewidth=0.8)
|
||
ax.set_title(f"Strategy Drawdown (%) | {period_label}")
|
||
ax.set_ylabel("Drawdown (%)")
|
||
ax.set_xlabel("Date")
|
||
apply_dark_style(fig, ax)
|
||
plt.tight_layout()
|
||
return fig
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Trade log
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _trade_df(trades: list[Trade]) -> pd.DataFrame:
|
||
rows = []
|
||
for t in trades:
|
||
tag = "Whipsaw" if t.is_whipsaw else ("Force-Close" if t.forced_exit else "-")
|
||
rows.append({
|
||
"Entry Date": t.entry_date.date(),
|
||
"Exit Date": t.exit_date.date() if t.exit_date else "-",
|
||
"Entry ($)": round(t.entry_price, 2),
|
||
"Exit ($)": round(t.exit_price, 2) if t.exit_price else "-",
|
||
"Mode": t.entry_mode,
|
||
"Hold (days)": t.holding_days,
|
||
"Return (%)": round(t.return_pct, 2),
|
||
"Cost ($)": round(t.cost_buy + t.cost_sell, 2),
|
||
"Net P&L ($)": round(t.net_pnl, 2),
|
||
"Flag": tag,
|
||
})
|
||
return pd.DataFrame(rows)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Assumptions text
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
_ASSUMPTIONS_MD = """
|
||
**信号规则（与策略仪表盘 Step 2 一致）**
|
||
- **A1** QQQ 日收盘 > MA200
|
||
- **A2** MA200 过去 20 日斜率 > −0.5%（平稳或向上）
|
||
- **A3** 连续 ≥ 3 日收盘站上 MA200
|
||
|
||
**仓位比例（Step 3，宏观评分 = 3）**
|
||
- 距突破价 ≤ 6% → Direct Buy → 当前现金 **2/3**
|
||
- 距突破价 > 6% → Split → 当前现金 **1/3**
|
||
|
||
**初始入场** 若回测开始时 QQQ 已在 MA200 之上（signal=True），立即入场
|
||
|
||
**离场** QQQ 连续 2 日收盘低于 MA200
|
||
|
||
**成本** 每笔买/卖各 $1.99（富途美股：$0.99 佣金 + $1.00 平台费）
|
||
|
||
**基准** QQQ 买入持有（同期，等额资金）
|
||
|
||
**Whipsaw** 持仓 ≤ 10 天且亏损
|
||
"""
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Per-period sub-tab renderer
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _render_period(period_key: str) -> None:
|
||
"""Render one backtest period sub-tab (run button + cached results)."""
|
||
spec = _PERIOD_SPECS[period_key]
|
||
label = spec["label"]
|
||
cache_key = f"bt_{period_key}"
|
||
|
||
run = st.button("▶ 运行回测", type="primary", key=f"bt_run_{period_key}")
|
||
|
||
if not run and cache_key not in st.session_state:
|
||
st.info(f"点击「运行回测」开始计算 **{label}** 的回测。")
|
||
with st.expander("回测假设说明"):
|
||
st.markdown(_ASSUMPTIONS_MD)
|
||
return
|
||
|
||
# ── Run ───────────────────────────────────────────────────────────────
|
||
if run:
|
||
with st.spinner(f"下载 {label} 数据（含预热期）…"):
|
||
try:
|
||
qqq_df, tqqq_df = _fetch(period_key)
|
||
except Exception as exc:
|
||
st.error(f"数据下载失败：{exc}")
|
||
return
|
||
|
||
if qqq_df.empty or tqqq_df.empty:
|
||
st.error("QQQ 或 TQQQ 数据为空，请稍后重试。")
|
||
return
|
||
|
||
qqq_full = qqq_df["Close"].squeeze().dropna()
|
||
tqqq_full = tqqq_df["Close"].squeeze().dropna()
|
||
|
||
if len(qqq_full) < 220:
|
||
st.error(f"数据量不足（{len(qqq_full)} 天），无法计算 MA200。")
|
||
return
|
||
|
||
t_start = _get_effective_start(qqq_full, tqqq_full, period_key)
|
||
qqq_eff = qqq_full[t_start:]
|
||
|
||
with st.spinner("计算信号并模拟交易…"):
|
||
sig = _build_signals(qqq_full)
|
||
trades, eq = _simulate(sig, tqqq_full, trade_start=t_start)
|
||
m = _metrics(trades, eq, qqq_eff)
|
||
|
||
st.session_state[cache_key] = {
|
||
"trades": trades, "equity": eq,
|
||
"qqq_eff": qqq_eff, "metrics": m,
|
||
}
|
||
st.success("回测完成！")
|
||
|
||
# ── Render ────────────────────────────────────────────────────────────
|
||
res = st.session_state.get(cache_key)
|
||
if res is None:
|
||
return
|
||
|
||
trades = res["trades"]
|
||
equity = res["equity"]
|
||
qqq_eff = res["qqq_eff"]
|
||
m = res["metrics"]
|
||
|
||
qqq_r = m["qqq_return"]
|
||
bh_final = _INITIAL_CAPITAL * (1 + qqq_r / 100)
|
||
strat_r = m["strategy_return"]
|
||
delta = strat_r - qqq_r
|
||
|
||
st.caption(
|
||
f"数据区间：{qqq_eff.index[0].date()} → {qqq_eff.index[-1].date()}"
|
||
)
|
||
|
||
# ── Metric block 1: performance ───────────────────────────────────────
|
||
with st.container(border=True):
|
||
# Row 1: Total return
|
||
c1, c2, c3, c4 = st.columns(4)
|
||
c1.metric("LRS 总收益率", f"{strat_r:+.1f}%",
|
||
f"vs QQQ {delta:+.1f}%", delta_color="normal")
|
||
c2.metric("QQQ B&H 收益率", f"{qqq_r:+.1f}%")
|
||
c3.metric("LRS 最终资产", f"${m['final_value']:,.0f}")
|
||
c4.metric("QQQ B&H 最终", f"${bh_final:,.0f}")
|
||
|
||
st.divider()
|
||
|
||
# Row 2: Risk stats
|
||
c1, c2, c3, c4 = st.columns(4)
|
||
c1.metric("最大回撤", f"{m['max_dd']:.1f}%")
|
||
c2.metric("交易次数",
|
||
str(m["n_trades"]),
|
||
f"盈 {m['n_wins']} / 亏 {m['n_losses']}")
|
||
c3.metric("胜率", f"{m['win_rate']:.0f}%")
|
||
c4.metric("Whipsaw",
|
||
f"{m['n_whipsaws']} 次",
|
||
f"均持仓 {m['avg_days']:.0f} 天")
|
||
|
||
# ── Equity curve ──────────────────────────────────────────────────────
|
||
with st.container(border=True):
|
||
st.markdown("##### Equity Curve")
|
||
st.pyplot(_fig_equity(equity, qqq_eff, trades, label), clear_figure=True)
|
||
|
||
# ── Per-trade returns ─────────────────────────────────────────────────
|
||
with st.container(border=True):
|
||
st.markdown("##### Per-Trade Return")
|
||
st.pyplot(_fig_per_trade(trades, label), clear_figure=True)
|
||
|
||
# ── Drawdown ──────────────────────────────────────────────────────────
|
||
with st.container(border=True):
|
||
st.markdown("##### Drawdown")
|
||
st.pyplot(_fig_drawdown(equity, label), clear_figure=True)
|
||
|
||
# ── Trade log ─────────────────────────────────────────────────────────
|
||
with st.container(border=True):
|
||
st.markdown("##### 交易明细")
|
||
if trades:
|
||
st.dataframe(
|
||
_trade_df(trades),
|
||
use_container_width=True,
|
||
hide_index=True,
|
||
column_config={
|
||
"Return (%)": st.column_config.NumberColumn(format="%.2f %%"),
|
||
"Net P&L ($)": st.column_config.NumberColumn(format="$ %.2f"),
|
||
},
|
||
)
|
||
st.caption(
|
||
f"Total cost: ${m['total_cost']:.2f} | "
|
||
f"Period: {qqq_eff.index[0].date()} → {qqq_eff.index[-1].date()}"
|
||
)
|
||
else:
|
||
st.info("该区间内未产生任何交易信号。")
|
||
|
||
with st.expander("回测假设说明"):
|
||
st.markdown(_ASSUMPTIONS_MD)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Entry point
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def render_backtest_tab() -> None:
|
||
st.subheader("LRS TQQQ 策略回测")
|
||
st.caption(
|
||
"LRS 策略 vs QQQ 买入持有 | 六个时间窗口独立缓存，切换 tab 无需重新运行。"
|
||
)
|
||
|
||
tabs = st.tabs([_PERIOD_SPECS[k]["label"] for k in _PERIOD_ORDER])
|
||
for tab, period_key in zip(tabs, _PERIOD_ORDER):
|
||
with tab:
|
||
_render_period(period_key)
|