freedakgmail
239 lines
10 KiB
Python
239 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Top 3 最挣钱算法回测 v5（使用5分钟K线实时价格版）
|
||
|
||
对比三组数据:
|
||
1. 原始版(日线close/open) - 作为基准
|
||
2. 5分钟实时价格版 - 使用stock_kline_5min的10:00买入价、15:00卖出价
|
||
3. Mid价格版 - 使用(open+close)/2作为替代
|
||
|
||
Top 3 算法:
|
||
🏆 v4|触发≥2+止盈10+损8 → +53,100 (33.1%, 年化28.3%)
|
||
🥈 v3|止盈10+损8 → +49,500 (28.4%, 年化24.4%)
|
||
🥉 v4.2-K1|延迟2天+止盈10+损8 → +48,920 (28.1%, 年化24.1%)
|
||
"""
|
||
import sys
|
||
import os
|
||
import time
|
||
from datetime import date, datetime
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from backtest_recommend import (
|
||
get_db_conn, get_codes_with_data, run_backtest, START_DATE,
|
||
MAX_POSITION_AMOUNT, MAX_CONCURRENT_POSITIONS, PRICE_MIN, PRICE_MAX,
|
||
SELL_COOLDOWN_DAYS, MAX_BUYS_PER_DAY,
|
||
)
|
||
|
||
|
||
def main():
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description='Top 3 算法回测')
|
||
parser.add_argument('--start', type=str, default=None, help='回测起始日 YYYY-MM-DD（默认取扫描历史最早日期）')
|
||
parser.add_argument('--end', type=str, default=None, help='回测结束日 YYYY-MM-DD（默认今天）')
|
||
args = parser.parse_args()
|
||
|
||
conn = get_db_conn()
|
||
end = date.today()
|
||
if args.end:
|
||
end = datetime.strptime(args.end, '%Y-%m-%d').date()
|
||
|
||
# 使用完整扫描数据期间（而非 START_DATE 的短期）
|
||
with conn.cursor() as cur:
|
||
cur.execute("SELECT min(scan_date) FROM stock_scan_history")
|
||
first_scan = cur.fetchone()[0]
|
||
|
||
if args.start:
|
||
start = datetime.strptime(args.start, '%Y-%m-%d').date()
|
||
print(f"📅 回测起始: {start}（用户指定）")
|
||
else:
|
||
start = first_scan if first_scan else START_DATE
|
||
print(f"📅 回测起始: {start}（扫描历史最早日期）")
|
||
|
||
# 检查5分钟K线数据覆盖
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
SELECT count(*), count(DISTINCT code), count(DISTINCT dt::date),
|
||
min(dt::date), max(dt::date)
|
||
FROM stock_kline_5min
|
||
""")
|
||
cnt, codes, days, min_d, max_d = cur.fetchone()
|
||
print(f"📊 stock_kline_5min 数据: {cnt:,}条 | {codes}只股票 | {days}天 | {min_d}~{max_d}")
|
||
|
||
cur.execute("SELECT count(DISTINCT scan_date) FROM stock_scan_history WHERE scan_date >= %s", (start,))
|
||
scan_days = cur.fetchone()[0]
|
||
print(f"📊 stock_scan_history: {scan_days}天扫描数据")
|
||
|
||
cur.execute("SELECT count(DISTINCT trade_date) FROM stock_kline_daily WHERE trade_date >= %s", (start,))
|
||
kline_days = cur.fetchone()[0]
|
||
print(f"📊 stock_kline_daily: {kline_days}天K线数据")
|
||
|
||
# Top 3 算法配置
|
||
top3_algos = [
|
||
("🏆 v4|触发≥2+止盈10+损8", {
|
||
"min_buy_triggered": 2,
|
||
"take_profit_pct": 10, "stop_loss_pct": 8,
|
||
}),
|
||
("🥈 v3|止盈10+损8", {
|
||
"take_profit_pct": 10, "stop_loss_pct": 8,
|
||
}),
|
||
("🥉 v4.2-K1|延迟2天+止盈10+损8", {
|
||
"sell_confirm_days": 2,
|
||
"take_profit_pct": 10, "stop_loss_pct": 8,
|
||
}),
|
||
]
|
||
|
||
# 3 种定价模式
|
||
price_modes = [
|
||
("日线(原版)", False, '10:00', '15:00'), # 用daily open/close
|
||
("5分钟最优时点", True, '09:35', '13:40'), # v7: 最优时点
|
||
("5分钟旧时点", True, '10:00', '15:00'), # v5: 旧默认时点 (对比用)
|
||
]
|
||
|
||
print("\n" + "=" * 140)
|
||
print(" Top 3 最挣钱算法 × 3种定价模式 对比回测 v7")
|
||
print("=" * 140)
|
||
print(f" 回测区间: {start} ~ {end}")
|
||
print(f" 定价说明:")
|
||
print(f" 日线(原版): 买入用开盘价, 卖出用收盘价")
|
||
print(f" 5分钟最优时点: 买入用09:35实时价, 卖出用13:40实时价 (v7网格搜索最优)")
|
||
print(f" 5分钟旧时点: 买入用10:00实时价, 卖出用15:00实时价 (v5旧默认)")
|
||
print(f" 股价区间: {PRICE_MIN}~{PRICE_MAX} 元 | 每笔1000股 | 每日最多买{MAX_BUYS_PER_DAY}只")
|
||
print("=" * 140)
|
||
print()
|
||
|
||
rows = []
|
||
total_scenarios = len(top3_algos) * len(price_modes)
|
||
idx = 0
|
||
|
||
for algo_name, algo_params in top3_algos:
|
||
for mode_name, use_5min, bt, st in price_modes:
|
||
idx += 1
|
||
scenario_name = f"{algo_name} | {mode_name}"
|
||
print(f"[{idx}/{total_scenarios}] {scenario_name}", flush=True)
|
||
|
||
t0 = time.time()
|
||
result = run_backtest(
|
||
conn, start_date=start, end_date=end,
|
||
use_5min_prices=use_5min,
|
||
buy_time=bt, sell_time=st,
|
||
verbose=False, **algo_params
|
||
)
|
||
elapsed = time.time() - t0
|
||
|
||
if not result or not result.get('stats'):
|
||
print(f" ❌ 无结果")
|
||
rows.append((scenario_name, algo_name, mode_name, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
|
||
continue
|
||
|
||
s = result['stats']
|
||
row = (
|
||
scenario_name,
|
||
algo_name,
|
||
mode_name,
|
||
s['profit'], # 3: 盈亏
|
||
s.get('capital_pct', 0), # 4: 真实收益率
|
||
s.get('capital_ann_pct', 0), # 5: 真实年化
|
||
s['win_rate'], # 6: 胜率
|
||
s.get('max_drawdown_pct', 0), # 7: 回撤%
|
||
s['avg_hold_days'], # 8: 持仓天
|
||
s['profit_factor'], # 9: 盈亏比
|
||
s['trade_count'], # 10: 交易数
|
||
s.get('max_capital', 0), # 11: 最大占用
|
||
s.get('5min_hit', 0), # 12: 5min命中
|
||
s.get('5min_miss', 0), # 13: 5min缺失
|
||
s.get('5min_coverage', 0), # 14: 5min覆盖率
|
||
)
|
||
rows.append(row)
|
||
mc = s.get('max_capital', 0)
|
||
cp = s.get('capital_pct', 0)
|
||
ca = s.get('capital_ann_pct', 0)
|
||
cov = s.get('5min_coverage', 0)
|
||
print(f" → ¥{s['profit']:>+10,.0f} 占用¥{mc:>8,.0f} "
|
||
f"真实{cp:>+6.1f}% 年化{ca:>+6.1f}% "
|
||
f"胜率{s['win_rate']:>5.1f}% "
|
||
f"{'5min覆盖' + str(cov) + '%' if use_5min else '日线'} "
|
||
f"({elapsed:.1f}s)", flush=True)
|
||
|
||
conn.close()
|
||
|
||
# ═══ 输出对比表 ═══
|
||
print("\n\n" + "=" * 160)
|
||
print(" 📊 Top 3 算法 × 定价模式 完整对比表")
|
||
print("=" * 160)
|
||
header = (f"{'场景':<55} {'盈亏(元)':>10} {'占用资金':>10} {'真实收益':>8} {'真实年化':>8} "
|
||
f"{'胜率':>6} {'回撤%':>7} {'持仓':>5} {'盈亏比':>6} {'交易':>5} {'5min':>6}")
|
||
print(header)
|
||
print("-" * 160)
|
||
|
||
prev_algo = None
|
||
for row in rows:
|
||
name = row[0]
|
||
algo = row[1]
|
||
mode = row[2]
|
||
profit, cp, ca = row[3], row[4], row[5]
|
||
wr, dd = row[6], row[7]
|
||
hold, pf, n, mc = row[8], row[9], row[10], row[11]
|
||
cov = row[14]
|
||
|
||
if prev_algo and prev_algo != algo:
|
||
print("-" * 160)
|
||
prev_algo = algo
|
||
|
||
cov_str = f"{cov:.0f}%" if cov > 0 else "日线"
|
||
print(f"{name:<55} {profit:>+10,.0f} {'¥'+str(int(mc)):>10} {cp:>+7.1f}% {ca:>+7.1f}% "
|
||
f"{wr:>5.1f}% {dd:>6.1f}% {hold:>4.0f}天 {pf:>6.2f} {n:>5} {cov_str:>6}")
|
||
|
||
print("=" * 160)
|
||
|
||
# ═══ 算法级汇总 ═══
|
||
print("\n 📊 按算法汇总:")
|
||
for algo_name, _ in top3_algos:
|
||
algo_rows = [r for r in rows if r[1] == algo_name]
|
||
if len(algo_rows) >= 2:
|
||
baseline = algo_rows[0] # 日线版
|
||
realtime = algo_rows[1] # 5分钟版
|
||
|
||
diff_profit = realtime[3] - baseline[3]
|
||
diff_ann = realtime[5] - baseline[5]
|
||
diff_wr = realtime[6] - baseline[6]
|
||
|
||
print(f"\n {algo_name}:")
|
||
print(f" 日线(原版) : 盈亏 ¥{baseline[3]:>+10,.0f} 年化 {baseline[5]:>+6.1f}% 胜率 {baseline[6]:.1f}%")
|
||
print(f" 5分钟实时 : 盈亏 ¥{realtime[3]:>+10,.0f} 年化 {realtime[5]:>+6.1f}% 胜率 {realtime[6]:.1f}% "
|
||
f"(5min覆盖{realtime[14]:.0f}%)")
|
||
icon = '📈' if diff_profit > 0 else ('📉' if diff_profit < 0 else '➖')
|
||
print(f" {icon} 差异: 盈亏{diff_profit:>+,.0f} 年化{diff_ann:>+.1f}% 胜率{diff_wr:>+.1f}%")
|
||
|
||
# ═══ 写入文件 ═══
|
||
out_dir = os.path.join(os.path.dirname(__file__), 'docs')
|
||
os.makedirs(out_dir, exist_ok=True)
|
||
out_path = os.path.join(out_dir, 'backtest_top3_5min.md')
|
||
with open(out_path, 'w', encoding='utf-8') as f:
|
||
f.write("# Top 3 算法 × 5分钟实时价格 回测对比\n\n")
|
||
f.write(f"回测区间: {start} ~ {end}\n\n")
|
||
f.write("## 定价模式\n\n")
|
||
f.write("| 模式 | 买入价 | 卖出价 | 说明 |\n")
|
||
f.write("|------|--------|--------|------|\n")
|
||
f.write("| 日线(原版) | 当日开盘价 | 当日收盘价 | 原始基准 |\n")
|
||
f.write("| 5分钟实时 | 10:00 5min收盘 | 15:00 5min收盘 | 有5min数据用5min, 无则用mid=(开盘+收盘)/2 |\n\n")
|
||
f.write("## 对比结果\n\n")
|
||
f.write("| 算法 | 定价 | 盈亏(元) | 占用资金 | 真实收益 | 真实年化 | 胜率 | 回撤% | 盈亏比 | 交易 | 5min覆盖 |\n")
|
||
f.write("|------|------|---------|---------|---------|---------|------|-------|--------|------|----------|\n")
|
||
for row in rows:
|
||
name, algo, mode = row[0], row[1], row[2]
|
||
profit, cp, ca = row[3], row[4], row[5]
|
||
wr, dd = row[6], row[7]
|
||
hold, pf, n, mc = row[8], row[9], row[10], row[11]
|
||
cov = row[14]
|
||
cov_str = f"{cov:.0f}%" if cov > 0 else "-"
|
||
f.write(f"| {algo} | {mode} | {profit:+,.0f} | ¥{mc:,.0f} | {cp:+.1f}% | {ca:+.1f}% | "
|
||
f"{wr:.1f}% | {dd:.1f}% | {pf:.2f} | {n} | {cov_str} |\n")
|
||
f.write("\n")
|
||
|
||
print(f"\n结果已写入 {out_path}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|