107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
"""
|
|
evolution/lesson_extractor.py — 交易后教训提取
|
|
分析已平仓交易，用LLM提取"为什么赢/亏"的教训
|
|
"""
|
|
import sys, os, json, sqlite3
|
|
from datetime import datetime, timedelta
|
|
|
|
sys.path.insert(0, '/home/hmo/MoFin')
|
|
sys.path.insert(0, '/home/hmo/MoFin/deploy/profile-scripts')
|
|
|
|
DB = os.environ.get('MOFIN_DB', '/home/hmo/MoFin/data/mofin.db')
|
|
|
|
|
|
def get_closed_trades(conn, days=30):
|
|
"""取近N天已平仓的交易（从strategy_research的trades里取最近的，或从holding_strategies推断）"""
|
|
# 从最近的回测结果取交易（作为样本分析）
|
|
r = conn.execute("""
|
|
SELECT results_json FROM strategy_research
|
|
WHERE version='v_next4' AND period_tag='5y' ORDER BY id DESC LIMIT 1
|
|
""").fetchone()
|
|
if not r:
|
|
return []
|
|
res = json.loads(r[0])
|
|
trades = res.get('trades', [])
|
|
# 按日期排序，取最近的
|
|
recent = sorted(trades, key=lambda x: x.get('entry_date', ''), reverse=True)[:10]
|
|
return recent
|
|
|
|
|
|
def analyze_trade(trade):
|
|
"""分析单笔交易的成败原因（规则化，非LLM）"""
|
|
profit = trade.get('profit_pct', 0)
|
|
hold_days = trade.get('hold_days', 0)
|
|
factors = trade.get('factors', {})
|
|
|
|
lessons = []
|
|
|
|
# 盈利交易的共性
|
|
if profit > 15:
|
|
if factors.get('mkt_adx', 0) > 25:
|
|
lessons.append(('win_pattern', f"大盘趋势强(ADX={factors['mkt_adx']:.0f})时盈利{profit:.1f}%", 0.8))
|
|
if factors.get('sector_above_ma20'):
|
|
lessons.append(('win_pattern', f"板块在MA20上方时盈利{profit:.1f}%", 0.7))
|
|
if trade.get('dna'):
|
|
lessons.append(('win_pattern', f"动量基因(DNA)票盈利{profit:.1f}%", 0.9))
|
|
|
|
# 亏损交易的共性
|
|
if profit < -5:
|
|
if factors.get('mkt_slope', 0) < -0.5:
|
|
lessons.append(('loss_pattern', f"大盘斜率负({factors['mkt_slope']:.2f})时亏损{profit:.1f}%", 0.7))
|
|
if not factors.get('sector_above_ma20'):
|
|
lessons.append(('loss_pattern', f"板块在MA20下方时亏损{profit:.1f}%", 0.6))
|
|
if hold_days < 5:
|
|
lessons.append(('loss_pattern', f"持仓{hold_days}天短于5天时亏损{profit:.1f}%", 0.5))
|
|
|
|
# 长持盈利
|
|
if profit > 10 and hold_days > 30:
|
|
lessons.append(('win_pattern', f"长持{hold_days}天盈利{profit:.1f}%", 0.85))
|
|
|
|
return lessons
|
|
|
|
|
|
def extract_lessons(days=30):
|
|
"""提取近N天交易的教训"""
|
|
conn = sqlite3.connect(DB)
|
|
conn.row_factory = sqlite3.Row
|
|
|
|
trades = get_closed_trades(conn, days)
|
|
if not trades:
|
|
print("无交易数据", flush=True)
|
|
conn.close()
|
|
return []
|
|
|
|
all_lessons = []
|
|
for t in trades:
|
|
lessons = analyze_trade(t)
|
|
for lesson_type, text, confidence in lessons:
|
|
all_lessons.append({
|
|
'strategy_version': 'v_next4',
|
|
'trade_id': t.get('id', 0),
|
|
'lesson_type': lesson_type,
|
|
'lesson_text': text,
|
|
'confidence': confidence,
|
|
'profit_pct': t.get('profit_pct', 0),
|
|
'entry_date': t.get('entry_date', ''),
|
|
})
|
|
|
|
# 写入 strategy_lessons 表
|
|
for l in all_lessons:
|
|
conn.execute("""
|
|
INSERT INTO strategy_lessons (strategy_version, trade_id, lesson_type, lesson_text, confidence, applied)
|
|
VALUES (?, ?, ?, ?, ?, 0)
|
|
""", (l['strategy_version'], l['trade_id'], l['lesson_type'], l['lesson_text'], l['confidence']))
|
|
conn.commit()
|
|
|
|
# 打印
|
|
print(f"分析 {len(trades)} 笔交易, 提取 {len(all_lessons)} 条教训", flush=True)
|
|
for l in all_lessons[:5]:
|
|
print(f" [{l['lesson_type']}] {l['lesson_text']} (置信度{l['confidence']})", flush=True)
|
|
|
|
conn.close()
|
|
return all_lessons
|
|
|
|
|
|
if __name__ == '__main__':
|
|
extract_lessons()
|