Files
stock/stock-html/run_backtest_quarterly.py
T
freedakgmail 9c7d7abdd4 Initial commit
2026-07-17 18:49:35 +08:00

491 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
20万本金 × 按季度投资 × 多算法对比回测 (v7.0 最优交易时点)
条件(v7.0: 最优时点 + 完全无人为限制):
- 本金:¥200,000(唯一约束)
- 单只上限:无
- 最大持仓:无
- 每笔股数:动态(总资金 × position_pct%,每笔等金额)
- 股价区间:无
- 每日最多买入:无
- 买入时间:09:35(最优,网格搜索验证)
- 卖出时间:13:40(最优,网格搜索验证)
- 回测区间:2025-01-01 ~ 最新,按季度分段
"""
import sys
import os
import time
from datetime import date, datetime
from collections import defaultdict
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from backtest_recommend import (
get_db_conn, run_backtest, START_DATE, SELL_COOLDOWN_DAYS,
)
# ─── 20万本金配置(v5.2: 完全无人为限制 + 动态仓位) ─────────────────
CAPITAL = 200_000 # 总本金(唯一约束)
POSITION_PCT = 5 # 单笔仓位 = 总资金的5%(¥10,000/笔,约可持20只)
# ─── 季度定义 ─────────────────────
QUARTERS = [
("2025-Q1", date(2025, 1, 2), date(2025, 3, 31)),
("2025-Q2", date(2025, 4, 1), date(2025, 6, 30)),
("2025-Q3", date(2025, 7, 1), date(2025, 9, 30)),
("2025-Q4", date(2025, 10, 1), date(2025, 12, 31)),
("2026-Q1", date(2026, 1, 5), date(2026, 2, 25)),
# 完整区间
("全期间", date(2025, 1, 2), date(2026, 2, 25)),
]
# ─── 算法配置 ─────────────────────
ALGORITHMS = [
# 名称, 参数字典
("v3|基线(TP10+SL8)", {
"take_profit_pct": 10, "stop_loss_pct": 8,
}),
("v4|触发≥2+TP10+SL8", {
"min_buy_triggered": 2, "take_profit_pct": 10, "stop_loss_pct": 8,
}),
("v4.2|延迟2天+TP10+SL8", {
"sell_confirm_days": 2, "take_profit_pct": 10, "stop_loss_pct": 8,
}),
("v4|忽略卖出+TP10+SL8", {
"ignore_sell_signal": True, "take_profit_pct": 10, "stop_loss_pct": 8,
}),
("v4|触发≥2+TP15+SL8", {
"min_buy_triggered": 2, "take_profit_pct": 15, "stop_loss_pct": 8,
}),
("v4|触发≥2+TP10+SL5", {
"min_buy_triggered": 2, "take_profit_pct": 10, "stop_loss_pct": 5,
}),
("v4.2|延迟2天+触发≥2+TP10+SL8", {
"sell_confirm_days": 2, "min_buy_triggered": 2,
"take_profit_pct": 10, "stop_loss_pct": 8,
}),
("v4.1|跟踪止盈8/3+SL5", {
"ignore_sell_signal": True, "trailing_start_pct": 8, "trailing_gap_pct": 3,
"stop_loss_pct": 5,
}),
("v4.1|跟踪止盈8/3+触发≥2+SL5", {
"ignore_sell_signal": True, "trailing_start_pct": 8, "trailing_gap_pct": 3,
"stop_loss_pct": 5, "min_buy_triggered": 2,
}),
("v4|触发≥2+TP10+SL8+持仓≤30天", {
"min_buy_triggered": 2, "take_profit_pct": 10, "stop_loss_pct": 8,
"max_hold_days": 30,
}),
# ── v5.2: 信号加权仓位 (强信号1.5倍/较强1.2倍) ──
("v5.2|延迟2天+触发≥2+TP10+SL8+信号加权", {
"sell_confirm_days": 2, "min_buy_triggered": 2,
"take_profit_pct": 10, "stop_loss_pct": 8,
"signal_weight": True,
}),
("v5.2|忽略卖出+TP10+SL8+信号加权", {
"ignore_sell_signal": True, "take_profit_pct": 10, "stop_loss_pct": 8,
"signal_weight": True,
}),
("v5.2|跟踪止盈8/3+SL5+信号加权", {
"ignore_sell_signal": True, "trailing_start_pct": 8, "trailing_gap_pct": 3,
"stop_loss_pct": 5, "signal_weight": True,
}),
]
def main():
conn = get_db_conn()
# 检查数据覆盖
with conn.cursor() as cur:
cur.execute("SELECT min(scan_date), max(scan_date), count(DISTINCT scan_date) FROM stock_scan_history")
scan_min, scan_max, scan_days = cur.fetchone()
print(f"📊 扫描数据: {scan_min} ~ {scan_max} ({scan_days}天)")
cur.execute("SELECT min(trade_date), max(trade_date), count(DISTINCT trade_date) FROM stock_kline_daily")
kline_min, kline_max, kline_days = cur.fetchone()
print(f"📊 K线数据: {kline_min} ~ {kline_max} ({kline_days}天)")
print(f"\n{'='*120}")
print(f" 💰 20万本金 按季度投资 × {len(ALGORITHMS)}种算法 对比回测 (v7.0 最优时点)")
print(f"{'='*120}")
print(f" 本金: ¥{CAPITAL:,} (唯一约束) | 单笔仓位: {POSITION_PCT}%=¥{int(CAPITAL*POSITION_PCT/100):,}/笔(动态)")
print(f" 限制: 单只上限=无 | 最大持仓=无 | 股价区间=无 | 每日买入=无 | 冷却期: {SELL_COOLDOWN_DAYS}")
print(f" 时点: 买入@09:35 | 卖出@13:40 (v7.0 网格搜索最优)")
print(f"{'='*120}\n")
# results[quarter_name][algo_name] = stats_dict
results = {}
total_runs = len(QUARTERS) * len(ALGORITHMS)
run_idx = 0
for q_name, q_start, q_end in QUARTERS:
results[q_name] = {}
for algo_name, algo_params in ALGORITHMS:
run_idx += 1
print(f" [{run_idx}/{total_runs}] {q_name} | {algo_name}", end="", flush=True)
t0 = time.time()
result = run_backtest(
conn,
start_date=q_start, end_date=q_end,
total_capital=CAPITAL, # v5.1: 总资金约束模式
position_pct=POSITION_PCT, # v5.2: 动态仓位(每笔=总资金×5%)
use_5min_prices=True, # v7: 使用5分钟实时价格
buy_time='09:35', # v7: 最优买入时间
sell_time='13:40', # v7: 最优卖出时间
verbose=False,
**algo_params,
)
elapsed = time.time() - t0
if result and result.get('stats'):
s = result['stats']
results[q_name][algo_name] = s
print(f" → ¥{s['profit']:>+10,.0f} 收益{s['capital_pct']:>+6.1f}% "
f"年化{s['capital_ann_pct']:>+7.1f}% 胜率{s['win_rate']:>5.1f}% "
f"({elapsed:.1f}s)", flush=True)
else:
results[q_name][algo_name] = None
print(f" → 无数据 ({elapsed:.1f}s)", flush=True)
conn.close()
# ─── 输出结果 ─────────────────────
# 1. 控制台大表
print(f"\n{'='*160}")
print(f" 📊 20万本金 × 按季度投资 完整对比表 (v5.2 动态仓位: 每笔={POSITION_PCT}%=¥{int(CAPITAL*POSITION_PCT/100):,})")
print(f"{'='*160}")
# 表头
header = f"{'算法':<36}"
for q_name, _, _ in QUARTERS:
header += f" | {q_name:>14}"
print(header)
print("-" * 160)
# 盈亏行
print("\n 📈 盈亏(元):")
print("-" * 160)
for algo_name, _ in ALGORITHMS:
row = f" {algo_name:<34}"
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
row += f" | {s['profit']:>+13,.0f}"
else:
row += f" | {'N/A':>13}"
print(row)
# 真实收益率行
print(f"\n 📊 真实收益率(%):")
print("-" * 160)
for algo_name, _ in ALGORITHMS:
row = f" {algo_name:<34}"
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
row += f" | {s['capital_pct']:>+12.1f}%"
else:
row += f" | {'N/A':>13}"
print(row)
# 年化收益率行
print(f"\n 📊 年化收益率(%):")
print("-" * 160)
for algo_name, _ in ALGORITHMS:
row = f" {algo_name:<34}"
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
method = s.get('capital_ann_method', '?')
tag = '(S)' if method == 'simple' else '(C)'
row += f" | {s['capital_ann_pct']:>+9.1f}%{tag}"
else:
row += f" | {'N/A':>13}"
print(row)
# 胜率行
print(f"\n 📊 胜率(%):")
print("-" * 160)
for algo_name, _ in ALGORITHMS:
row = f" {algo_name:<34}"
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
row += f" | {s['win_rate']:>12.1f}%"
else:
row += f" | {'N/A':>13}"
print(row)
# 盈亏比行
print(f"\n 📊 盈亏比:")
print("-" * 160)
for algo_name, _ in ALGORITHMS:
row = f" {algo_name:<34}"
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
row += f" | {s['profit_factor']:>13.2f}"
else:
row += f" | {'N/A':>13}"
print(row)
# 交易笔数行
print(f"\n 📊 交易笔数:")
print("-" * 160)
for algo_name, _ in ALGORITHMS:
row = f" {algo_name:<34}"
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
row += f" | {s['trade_count']:>13}"
else:
row += f" | {'N/A':>13}"
print(row)
print(f"\n{'='*160}")
# ─── 找出各季度最优算法 ─────────
print(f"\n 🏆 各季度最优算法:")
print("-" * 100)
for q_name, _, _ in QUARTERS:
best_algo = None
best_profit = -float('inf')
best_ann = -float('inf')
for algo_name, _ in ALGORITHMS:
s = results[q_name].get(algo_name)
if s and s['profit'] > best_profit:
best_profit = s['profit']
best_ann = s['capital_ann_pct']
best_algo = algo_name
best_stats = s
if best_algo:
print(f" {q_name:<14} → 🏆 {best_algo:<36} "
f"盈利 ¥{best_profit:>+10,.0f} 收益{best_stats['capital_pct']:>+6.1f}% "
f"年化{best_ann:>+7.1f}% 胜率{best_stats['win_rate']:.1f}% "
f"盈亏比{best_stats['profit_factor']:.2f}")
else:
print(f" {q_name:<14} → 无数据")
# ─── 输出到 Markdown ─────────
out_path = os.path.join(os.path.dirname(__file__), "docs", "backtest_quarterly_200k.md")
os.makedirs(os.path.dirname(out_path), exist_ok=True)
with open(out_path, "w", encoding="utf-8") as f:
f.write("# 💰 20万本金 × 按季度投资 × 多算法对比回测\n\n")
f.write(f"> 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n")
f.write("## 回测配置 (v5.2 动态仓位)\n\n")
f.write(f"| 参数 | 值 |\n|------|----|\n")
f.write(f"| 本金 | ¥{CAPITAL:,} (唯一约束) |\n")
f.write(f"| 单只上限 | 无(受总资金约束) |\n")
f.write(f"| 最大持仓 | 无(受总资金约束) |\n")
f.write(f"| 每笔仓位 | 动态: 总资金×{POSITION_PCT}% = ¥{int(CAPITAL*POSITION_PCT/100):,}/笔 |\n")
f.write(f"| 每笔股数 | 动态(根据股价自动计算,取整到100股) |\n")
f.write(f"| 股价区间 | 无 |\n")
f.write(f"| 每日最多买入 | 无 |\n")
f.write(f"| 冷却期 | {SELL_COOLDOWN_DAYS}天 |\n")
f.write(f"| 年化方法 | <90天用简单(S),≥90天用复利CAGR(C) |\n\n")
# 算法说明
f.write("## 算法说明\n\n")
f.write("| # | 算法 | 参数说明 |\n|---|------|--------|\n")
for i, (name, params) in enumerate(ALGORITHMS, 1):
param_str = ", ".join(f"{k}={v}" for k, v in params.items())
f.write(f"| {i} | {name} | {param_str} |\n")
f.write("\n")
# 盈亏对比表
f.write("## 一、盈亏对比(元)\n\n")
f.write(f"| 算法 |")
for q_name, _, _ in QUARTERS:
f.write(f" {q_name} |")
f.write("\n|------|")
for _ in QUARTERS:
f.write("--------|")
f.write("\n")
for algo_name, _ in ALGORITHMS:
f.write(f"| {algo_name} |")
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
val = f"{s['profit']:+,.0f}"
f.write(f" {val} |")
else:
f.write(" N/A |")
f.write("\n")
f.write("\n")
# 真实收益率
f.write("## 二、真实收益率(%)\n\n")
f.write(f"| 算法 |")
for q_name, _, _ in QUARTERS:
f.write(f" {q_name} |")
f.write("\n|------|")
for _ in QUARTERS:
f.write("--------|")
f.write("\n")
for algo_name, _ in ALGORITHMS:
f.write(f"| {algo_name} |")
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
f.write(f" {s['capital_pct']:+.1f}% |")
else:
f.write(" N/A |")
f.write("\n")
f.write("\n")
# 年化收益率
f.write("## 三、年化收益率(%)\n\n")
f.write("> (S)=简单年化(<90天)(C)=复利CAGR(≥90天)\n\n")
f.write(f"| 算法 |")
for q_name, _, _ in QUARTERS:
f.write(f" {q_name} |")
f.write("\n|------|")
for _ in QUARTERS:
f.write("--------|")
f.write("\n")
for algo_name, _ in ALGORITHMS:
f.write(f"| {algo_name} |")
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
method = s.get('capital_ann_method', '?')
tag = '(S)' if method == 'simple' else '(C)'
f.write(f" {s['capital_ann_pct']:+.1f}%{tag} |")
else:
f.write(" N/A |")
f.write("\n")
f.write("\n")
# 胜率
f.write("## 四、胜率(%)\n\n")
f.write(f"| 算法 |")
for q_name, _, _ in QUARTERS:
f.write(f" {q_name} |")
f.write("\n|------|")
for _ in QUARTERS:
f.write("--------|")
f.write("\n")
for algo_name, _ in ALGORITHMS:
f.write(f"| {algo_name} |")
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
f.write(f" {s['win_rate']:.1f}% |")
else:
f.write(" N/A |")
f.write("\n")
f.write("\n")
# 盈亏比
f.write("## 五、盈亏比\n\n")
f.write(f"| 算法 |")
for q_name, _, _ in QUARTERS:
f.write(f" {q_name} |")
f.write("\n|------|")
for _ in QUARTERS:
f.write("--------|")
f.write("\n")
for algo_name, _ in ALGORITHMS:
f.write(f"| {algo_name} |")
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
f.write(f" {s['profit_factor']:.2f} |")
else:
f.write(" N/A |")
f.write("\n")
f.write("\n")
# 最大占用资金
f.write("## 六、最大占用资金(元)\n\n")
f.write(f"| 算法 |")
for q_name, _, _ in QUARTERS:
f.write(f" {q_name} |")
f.write("\n|------|")
for _ in QUARTERS:
f.write("--------|")
f.write("\n")
for algo_name, _ in ALGORITHMS:
f.write(f"| {algo_name} |")
for q_name, _, _ in QUARTERS:
s = results[q_name].get(algo_name)
if s:
f.write(f" ¥{s['max_capital']:,.0f} |")
else:
f.write(" N/A |")
f.write("\n")
f.write("\n")
# 各季度最优算法
f.write("## 七、🏆 各季度最优算法\n\n")
f.write("| 季度 | 最优算法 | 盈利(元) | 真实收益 | 年化 | 胜率 | 盈亏比 |\n")
f.write("|------|---------|---------|---------|------|------|--------|\n")
for q_name, _, _ in QUARTERS:
best_algo = None
best_profit = -float('inf')
for algo_name, _ in ALGORITHMS:
s = results[q_name].get(algo_name)
if s and s['profit'] > best_profit:
best_profit = s['profit']
best_algo = algo_name
best_s = s
if best_algo:
method = best_s.get('capital_ann_method', '?')
tag = '(S)' if method == 'simple' else '(C)'
f.write(f"| {q_name} | **{best_algo}** | {best_profit:+,.0f} | "
f"{best_s['capital_pct']:+.1f}% | {best_s['capital_ann_pct']:+.1f}%{tag} | "
f"{best_s['win_rate']:.1f}% | {best_s['profit_factor']:.2f} |\n")
else:
f.write(f"| {q_name} | N/A | - | - | - | - | - |\n")
f.write("\n")
# 算法总盈利排名
f.write("## 八、算法全期间总收益排名\n\n")
algo_totals = []
for algo_name, _ in ALGORITHMS:
s = results["全期间"].get(algo_name)
if s:
algo_totals.append((algo_name, s))
algo_totals.sort(key=lambda x: x[1]['profit'], reverse=True)
f.write("| 排名 | 算法 | 全期间盈利 | 真实收益 | 年化(CAGR) | 胜率 | 盈亏比 | 最大回撤 | 占用资金 |\n")
f.write("|------|------|----------|---------|-----------|------|--------|---------|--------|\n")
for rank, (algo_name, s) in enumerate(algo_totals, 1):
medal = "🏆" if rank == 1 else ("🥈" if rank == 2 else ("🥉" if rank == 3 else f"#{rank}"))
f.write(f"| {medal} | {algo_name} | {s['profit']:+,.0f} | {s['capital_pct']:+.1f}% | "
f"{s['capital_ann_pct']:+.1f}% | {s['win_rate']:.1f}% | {s['profit_factor']:.2f} | "
f"{s['max_drawdown_pct']:.1f}% | ¥{s['max_capital']:,.0f} |\n")
f.write("\n")
# 分析结论
f.write("## 九、分析结论\n\n")
if algo_totals:
best_name, best_s = algo_totals[0]
f.write(f"### 🏆 全期间最优算法: {best_name}\n\n")
f.write(f"- 总盈利: **¥{best_s['profit']:+,.0f}**\n")
f.write(f"- 真实收益率: **{best_s['capital_pct']:+.1f}%**\n")
f.write(f"- 年化收益率: **{best_s['capital_ann_pct']:+.1f}%**\n")
f.write(f"- 胜率: **{best_s['win_rate']:.1f}%**\n")
f.write(f"- 盈亏比: **{best_s['profit_factor']:.2f}**\n")
f.write(f"- 最大回撤: **{best_s['max_drawdown_pct']:.1f}%**\n")
f.write(f"- 最大占用资金: **¥{best_s['max_capital']:,.0f}**{best_s['max_capital']/CAPITAL*100:.0f}%本金利用率)\n")
f.write(f"\n### 回报对比\n\n")
f.write(f"| 投资方式 | 年化收益 | 20万本金一年收益 |\n")
f.write(f"|---------|---------|----------------|\n")
f.write(f"| 银行定存 | 2.5% | ¥5,000 |\n")
f.write(f"| 余额宝 | 1.8% | ¥3,600 |\n")
f.write(f"| **本算法** | **{best_s['capital_ann_pct']:+.1f}%** | **¥{best_s['profit']:+,.0f}**(实际) |\n")
f.write(f"\n")
print(f"\n📝 结果已写入 {out_path}")
if __name__ == "__main__":
main()