""" 月度复盘引擎 — AI学习系统第二层 每月1号自动执行，分析维度（对应提示词七）: 1. 各信号类型近3个月胜率变化 2. 各币种近3个月表现排名 3. 当前trailing参数的出场效率 4. 连败暂停阈值是否合适 5. 量比过滤阈值是否合适 输出: 参数调整建议（在允许范围内） """ import logging from typing import Dict, List, Optional, Tuple from datetime import datetime, timezone from dataclasses import dataclass, field import pandas as pd import numpy as np from .memory import TradeMemory from ..config import settings as S logger = logging.getLogger(__name__) @dataclass class ParamSuggestion: """参数调整建议""" param_name: str current_value: float suggested_value: float min_allowed: float max_allowed: float reason: str sample_size: int confidence: float # 0-1 @dataclass class ReviewReport: """月度复盘报告""" month: str summary: dict signal_type_analysis: Dict[str, dict] coin_analysis: Dict[str, dict] trailing_analysis: dict streak_analysis: dict volume_analysis: dict suggestions: List[ParamSuggestion] market_phase: str class MonthlyReview: """ 月度复盘引擎 用法: review = MonthlyReview(memory) report = review.run('2024-10') review.print_report(report) """ def __init__(self, memory: TradeMemory): self.memory = memory def run(self, month: Optional[str] = None) -> ReviewReport: """ 执行月度复盘 Args: month: 复盘月份 'YYYY-MM'，默认当前月 Returns: ReviewReport """ if month is None: month = datetime.now(timezone.utc).strftime('%Y-%m') logger.info(f"Running monthly review for {month}") # 获取数据 all_trades = self.memory.get_all() recent_3m = self.memory.get_recent(months=3) if all_trades.empty: logger.warning("No trades in memory") return self._empty_report(month) # 1. 信号类型分析 signal_analysis = self._analyze_signal_types(recent_3m) # 2. 币种分析 coin_analysis = self._analyze_coins(recent_3m) # 3. Trailing出场效率 trailing_analysis = self._analyze_trailing(recent_3m) # 4. 连败分析 streak_analysis = self._analyze_streaks(recent_3m) # 5. 量比分析 volume_analysis = self._analyze_volume(recent_3m) # 6. 生成参数建议 suggestions = self._generate_suggestions( signal_analysis, trailing_analysis, streak_analysis, volume_analysis, recent_3m ) # 7. 市场阶段判断 market_phase = self._judge_market_phase(recent_3m) # 总结 summary = self.memory.calc_stats(recent_3m) return ReviewReport( month=month, summary=summary, signal_type_analysis=signal_analysis, coin_analysis=coin_analysis, trailing_analysis=trailing_analysis, streak_analysis=streak_analysis, volume_analysis=volume_analysis, suggestions=suggestions, market_phase=market_phase, ) # ==================== 分析维度 ==================== def _analyze_signal_types(self, df: pd.DataFrame) -> Dict[str, dict]: """各信号类型近3个月胜率变化""" result = {} for stype in ['共振', '日线', '4H', '做空']: grp = df[df['signal_type'] == stype] if grp.empty: continue stats = self.memory.calc_stats(grp) # 胜率趋势（按月拆分） if 'entry_dt' in grp.columns: grp = grp.copy() grp['month'] = pd.to_datetime(grp['entry_dt']).dt.to_period('M').astype(str) monthly_wr = {} for m, mg in grp.groupby('month'): wins = len(mg[mg['pnl_dollar'] > 0]) monthly_wr[m] = wins / len(mg) * 100 if len(mg) > 0 else 0 stats['monthly_win_rates'] = monthly_wr wr_values = list(monthly_wr.values()) if len(wr_values) >= 2: stats['wr_trend'] = wr_values[-1] - wr_values[0] else: stats['wr_trend'] = 0 result[stype] = stats return result def _analyze_coins(self, df: pd.DataFrame) -> Dict[str, dict]: """各币种近3个月表现排名""" return self.memory.calc_stats_by_group(df, 'coin') def _analyze_trailing(self, df: pd.DataFrame) -> dict: """Trailing出场效率分析""" trailing_trades = df[df['exit_reason'] == 'trailing'] if trailing_trades.empty: return {'n': 0, 'avg_efficiency': 0, 'suggestion': '样本不足'} by_coin_type = {} for coin in trailing_trades['coin'].unique(): coin_trades = trailing_trades[trailing_trades['coin'] == coin] avg_eff = coin_trades['exit_efficiency'].mean() by_coin_type[coin] = { 'n': len(coin_trades), 'avg_efficiency': avg_eff, 'avg_unrealized': coin_trades['max_unrealized_pct'].mean(), 'avg_realized': coin_trades['pnl_pct'].mean(), } overall_eff = trailing_trades['exit_efficiency'].mean() return { 'n': len(trailing_trades), 'avg_efficiency': overall_eff, 'by_coin': by_coin_type, 'optimal_note': '效率>0.7=trailing偏紧(留利润少), <0.4=偏松(回吐多)', } def _analyze_streaks(self, df: pd.DataFrame) -> dict: """连败分析""" # 只看做多 long_trades = df[df['direction'] == 'long'].sort_values('entry_dt') if long_trades.empty: return {'max_streak': 0, 'avg_streak': 0, 'streaks': []} streaks = [] current = 0 for _, t in long_trades.iterrows(): if t['pnl_dollar'] <= 0: current += 1 else: if current > 0: streaks.append(current) current = 0 if current > 0: streaks.append(current) return { 'max_streak': max(streaks) if streaks else 0, 'avg_streak': np.mean(streaks) if streaks else 0, 'streaks': streaks, 'streak_distribution': { '1-3': sum(1 for s in streaks if 1 <= s <= 3), '4-6': sum(1 for s in streaks if 4 <= s <= 6), '7+': sum(1 for s in streaks if s >= 7), }, 'current_threshold': S.CONSECUTIVE_LOSS_PAUSE, } def _analyze_volume(self, df: pd.DataFrame) -> dict: """量比过滤分析""" if 'breakout_vol_ratio' not in df.columns or df.empty: return {'n': 0} # 高量比 vs 正常量比的胜率对比 high_vol = df[df['breakout_vol_ratio'] > 2.0] normal_vol = df[df['breakout_vol_ratio'] <= 2.0] def wr(grp): if grp.empty: return 0 return len(grp[grp['pnl_dollar'] > 0]) / len(grp) * 100 return { 'n': len(df), 'high_vol_n': len(high_vol), 'high_vol_wr': wr(high_vol), 'normal_vol_n': len(normal_vol), 'normal_vol_wr': wr(normal_vol), 'avg_vol_ratio': df['breakout_vol_ratio'].mean(), } # ==================== 参数建议 ==================== def _generate_suggestions( self, signal_analysis: dict, trailing_analysis: dict, streak_analysis: dict, volume_analysis: dict, recent_df: pd.DataFrame, ) -> List[ParamSuggestion]: """ 基于分析结果生成参数调整建议 关键约束（提示词七）: - 每个参数只允许±20%浮动 - 最少N个样本才能做调整 - 每次只调一个参数 """ suggestions = [] # 1. Trailing参数建议 if trailing_analysis.get('n', 0) >= 20: avg_eff = trailing_analysis['avg_efficiency'] # 效率太低(< 0.4) → trailing偏松，建议收紧 # 效率太高(> 0.7) → trailing偏紧，建议放松 for coin, coin_data in trailing_analysis.get('by_coin', {}).items(): if coin_data['n'] < 5: continue eff = coin_data['avg_efficiency'] if coin == 'BTCUSDT': current = S.ZIGZAG_THRESHOLD_DAILY # 用作代理 current_trailing = 0.15 low, high = 0.12, 0.18 elif coin == 'ETHUSDT': current_trailing = 0.20 low, high = 0.16, 0.24 else: current_trailing = 0.25 low, high = 0.20, 0.30 if eff < 0.35: new_val = max(low, current_trailing - 0.02) suggestions.append(ParamSuggestion( param_name=f'{coin}_trailing', current_value=current_trailing, suggested_value=new_val, min_allowed=low, max_allowed=high, reason=f'出场效率{eff:.2f}偏低,回吐过多,建议收紧trailing', sample_size=coin_data['n'], confidence=min(coin_data['n'] / 20, 1.0), )) elif eff > 0.75: new_val = min(high, current_trailing + 0.02) suggestions.append(ParamSuggestion( param_name=f'{coin}_trailing', current_value=current_trailing, suggested_value=new_val, min_allowed=low, max_allowed=high, reason=f'出场效率{eff:.2f}偏高,可能过早离场,建议放宽trailing', sample_size=coin_data['n'], confidence=min(coin_data['n'] / 20, 1.0), )) # 2. 连败暂停阈值 if streak_analysis.get('streaks'): max_streak = streak_analysis['max_streak'] current_threshold = S.CONSECUTIVE_LOSS_PAUSE if max_streak >= current_threshold + 3: suggestions.append(ParamSuggestion( param_name='consecutive_loss_pause', current_value=current_threshold, suggested_value=min(7, current_threshold + 1), min_allowed=3, max_allowed=7, reason=f'最大连败{max_streak}笔,可适当放宽暂停阈值', sample_size=len(streak_analysis['streaks']), confidence=0.6, )) elif max_streak <= current_threshold - 2 and len(streak_analysis['streaks']) >= 5: suggestions.append(ParamSuggestion( param_name='consecutive_loss_pause', current_value=current_threshold, suggested_value=max(3, current_threshold - 1), min_allowed=3, max_allowed=7, reason=f'连败较少(最大{max_streak}),可收紧暂停阈值', sample_size=len(streak_analysis['streaks']), confidence=0.5, )) # 3. 4H信号权重（基于胜率） sig_4h = signal_analysis.get('4H', {}) if sig_4h.get('n', 0) >= 30: wr_4h = sig_4h['win_rate'] current_limit = S.POSITION_LIMIT.get('4H', 0.30) if wr_4h < 20: new_limit = max(0.15, current_limit - 0.05) suggestions.append(ParamSuggestion( param_name='4h_position_limit', current_value=current_limit, suggested_value=new_limit, min_allowed=0.15, max_allowed=0.35, reason=f'4H胜率{wr_4h:.0f}%偏低,建议降低仓位上限', sample_size=sig_4h['n'], confidence=0.7, )) elif wr_4h > 45: new_limit = min(0.35, current_limit + 0.05) suggestions.append(ParamSuggestion( param_name='4h_position_limit', current_value=current_limit, suggested_value=new_limit, min_allowed=0.15, max_allowed=0.35, reason=f'4H胜率{wr_4h:.0f}%良好,可适当提高仓位上限', sample_size=sig_4h['n'], confidence=0.6, )) return suggestions def _judge_market_phase(self, df: pd.DataFrame) -> str: """判断当前市场阶段""" if df.empty: return 'unknown' avg_btc_trend = df['btc_trend_60d'].mean() recent_wr = len(df[df['pnl_dollar'] > 0]) / len(df) * 100 if avg_btc_trend > 0.30 and recent_wr > 40: return '牛市中期（趋势强+胜率高）' elif avg_btc_trend > 0.10: return '牛市初期/复苏（趋势正）' elif avg_btc_trend > -0.10: return '震荡市（无明确方向）' elif avg_btc_trend > -0.30: return '熊市初期（趋势负）' else: return '深度熊市（趋势大幅负）' def _empty_report(self, month: str) -> ReviewReport: return ReviewReport( month=month, summary={'n': 0}, signal_type_analysis={}, coin_analysis={}, trailing_analysis={}, streak_analysis={}, volume_analysis={}, suggestions=[], market_phase='unknown', ) # ==================== 报告输出 ==================== def print_report(self, report: ReviewReport): """打印月度复盘报告""" print(f"\n{'='*65}") print(f" 浪浪AI 月度复盘 — {report.month}") print(f" 市场阶段: {report.market_phase}") print(f"{'='*65}") s = report.summary if s.get('n', 0) == 0: print(" 无交易数据") return print(f"\n 近3个月概览:") print(f" 交易: {s['n']}笔 胜率: {s['win_rate']:.1f}% " f"PF: {s['pf']:.2f} 净盈亏: ${s['total_pnl']:+,.0f}") # 信号类型 if report.signal_type_analysis: print(f"\n 信号类型分析:") print(f" {'类型':<8} {'交易':>4} {'胜率':>7} {'PF':>6} {'趋势'}") print(f" {'─'*40}") for stype, stats in report.signal_type_analysis.items(): trend = stats.get('wr_trend', 0) trend_arrow = '↑' if trend > 5 else '↓' if trend < -5 else '→' print(f" {stype:<8} {stats['n']:>4} {stats['win_rate']:>6.1f}% " f"{stats['pf']:>5.1f} {trend_arrow} {trend:+.0f}%") # 币种 if report.coin_analysis: print(f"\n 币种表现排名:") sorted_coins = sorted( report.coin_analysis.items(), key=lambda x: x[1].get('total_pnl', 0), reverse=True ) for coin, stats in sorted_coins[:8]: emoji = '🟢' if stats['total_pnl'] >= 0 else '🔴' print(f" {emoji} {coin.replace('USDT',''):<8} " f"{stats['n']:>3}笔 {stats['win_rate']:>5.1f}% " f"${stats['total_pnl']:>+10,.0f}") # Trailing效率 ta = report.trailing_analysis if ta.get('n', 0) > 0: print(f"\n Trailing出场效率: {ta['avg_efficiency']:.2f} " f"({ta['n']}笔)") # 连败 sa = report.streak_analysis if sa.get('streaks'): print(f"\n 连败分析: 最大{sa['max_streak']}笔, " f"平均{sa['avg_streak']:.1f}笔, " f"当前阈值{sa['current_threshold']}") # 参数建议 if report.suggestions: print(f"\n {'─'*65}") print(f" 参数调整建议 ({len(report.suggestions)}项):") for i, sug in enumerate(report.suggestions, 1): print(f"\n [{i}] {sug.param_name}") print(f" 当前: {sug.current_value:.3f} → 建议: {sug.suggested_value:.3f}") print(f" 范围: [{sug.min_allowed:.3f}, {sug.max_allowed:.3f}]") print(f" 理由: {sug.reason}") print(f" 样本: {sug.sample_size}笔, 置信度: {sug.confidence:.0%}") else: print(f"\n 参数建议: 暂无调整建议（参数表现正常）") print(f"\n{'='*65}") def to_dict(self, report: ReviewReport) -> dict: """将报告转为字典（用于LLM输入或JSON存储）""" return { 'month': report.month, 'market_phase': report.market_phase, 'summary': report.summary, 'signal_types': report.signal_type_analysis, 'coins': report.coin_analysis, 'trailing': report.trailing_analysis, 'streaks': report.streak_analysis, 'volume': report.volume_analysis, 'suggestions': [ { 'param': s.param_name, 'current': s.current_value, 'suggested': s.suggested_value, 'reason': s.reason, 'confidence': s.confidence, } for s in report.suggestions ], }