# -*- coding: utf-8 -*- """ factor_correlation.py — 因子拟合度分析 计算每个因子与以下指标的相关系数： 1. 同期价格（price_norm） 2. 未来N分钟价格变动（前瞻相关性，lag=1/3/5/10步） 3. 总分（total） 用法: python factor_correlation.py [--html] --html 额外生成 corr_report.html 可视化热力图 """ import re, json, math, sys, os from datetime import datetime _ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) LOG_FILE = os.path.join(_ROOT, "my_trader.log") OUT_HTML = os.path.join(_ROOT, "corr_report.html") MAX_POINTS = 2000 # 取最近N条，拉长时间周期 FACTOR_KEYS = ['TR','OB','TK','OI','FR','MP','VD','BTC','GM','IV','EX','LC','MR'] FACTOR_NAMES = { 'TR':'趋势','OB':'挂单','TK':'成交','OI':'持仓量', 'FR':'费率','MP':'痛点','VD':'量变','BTC':'大盘', 'GM':'Gamma','IV':'波动率','EX':'衰竭','LC':'清算','MR':'回归', } RE_PRICE = re.compile( r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}),\d+ \[MyTrader\] INFO: \[ETH\] \$([0-9,.]+) \|' ) RE_FACTOR = re.compile( r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}),\d+ \[MyTrader\] INFO: \[ETH\] ' r'TR:([+-]?\d+\.?\d*) OB:([+-]?\d+\.?\d*) FR:([+-]?\d+\.?\d*) ' r'TK:([+-]?\d+\.?\d*) OI:([+-]?\d+\.?\d*) MP:([+-]?\d+\.?\d*) ' r'VD:([+-]?\d+\.?\d*) BTC:([+-]?\d+\.?\d*) GM:([+-]?\d+\.?\d*) ' r'IV:([+-]?\d+\.?\d*) EX:([+-]?\d+\.?\d*) LC:([+-]?\d+\.?\d*) ' r'MR:([+-]?\d+\.?\d*) SM:([+-]?\d+\.?\d*) ' r'mom:[+-]?\d+\.?\d* flip:[+-]?\d+\.?\d* => ([+-]?\d+\.?\d*) -> \w+' ) def parse_log(path, max_pts): prices, rows = {}, [] try: lines = open(path, encoding='utf-8', errors='replace').readlines() except: return [] for line in lines: m = RE_PRICE.search(line) if m: prices[m.group(1)] = float(m.group(2).replace(',', '')) m = RE_FACTOR.search(line) if m: ts = m.group(1) rows.append({ 'ts': ts, 'price': prices.get(ts), 'TR': float(m.group(2)), 'OB': float(m.group(3)), 'FR': float(m.group(4)), 'TK': float(m.group(5)), 'OI': float(m.group(6)), 'MP': float(m.group(7)), 'VD': float(m.group(8)), 'BTC': float(m.group(9)), 'GM': float(m.group(10)), 'IV': float(m.group(11)), 'EX': float(m.group(12)), 'LC': float(m.group(13)), 'MR': float(m.group(14)), 'SM': float(m.group(15)), 'total': float(m.group(16)), }) # 填充价格 last_px = None for r in rows: if r['price'] is None: r['price'] = last_px else: last_px = r['price'] rows = [r for r in rows if r['price'] is not None] return rows[-max_pts:] def pearson(x, y): """皮尔逊相关系数""" n = len(x) if n < 3: return 0.0 mx, my = sum(x)/n, sum(y)/n num = sum((xi-mx)*(yi-my) for xi,yi in zip(x,y)) dx = math.sqrt(sum((xi-mx)**2 for xi in x)) dy = math.sqrt(sum((yi-my)**2 for yi in y)) if dx < 1e-10 or dy < 1e-10: return 0.0 return round(num/(dx*dy), 4) def compute_corr(rows, lags=(0, 1, 3, 5, 10)): """ 计算每个因子的相关系数矩阵 lag=0 → 与同期价格/总分相关 lag=N → 与N步后价格变动（对数收益）相关（前瞻性） """ n = len(rows) prices = [r['price'] for r in rows] # 价格归一化序列（用于同期相关） px_min, px_max = min(prices), max(prices) px_range = px_max - px_min if px_max != px_min else 1 px_mid = (px_max + px_min) / 2 price_norm = [(p - px_mid) / (px_range/2) for p in prices] results = {} for k in FACTOR_KEYS: factor_vals = [r[k] for r in rows] total_vals = [r['total'] for r in rows] res = {} # 与总分相关（同期） res['vs_total'] = pearson(factor_vals, total_vals) # 与同期归一化价格相关 res['vs_price_lag0'] = pearson(factor_vals, price_norm) # 与未来N步价格变动（对数收益）相关（前瞻性指标评估） for lag in lags: if lag == 0: continue if n <= lag: res[f'vs_price_lag{lag}'] = 0.0 continue # 未来lag步的对数收益 future_ret = [] fv_trimmed = [] for i in range(n - lag): p0 = prices[i] p1 = prices[i + lag] if p0 > 0 and p1 > 0: future_ret.append(math.log(p1/p0)) fv_trimmed.append(factor_vals[i]) res[f'vs_price_lag{lag}'] = pearson(fv_trimmed, future_ret) if future_ret else 0.0 results[k] = res return results def color_for(v): """相关系数着色：正相关绿，负相关红，接近0灰""" if v is None: return '#444' a = abs(v) if a < 0.05: return '#555' if v > 0: intensity = int(min(255, 50 + a * 205)) return f'#{0:02x}{intensity:02x}{0:02x}' else: intensity = int(min(255, 50 + a * 205)) return f'#{intensity:02x}{0:02x}{0:02x}' def text_color(bg): return '#fff' if bg in ('#555','#444') else '#000' if abs(int(bg[1:3],16)-128)<60 and abs(int(bg[3:5],16)-128)>60 else '#fff' def build_html_report(rows, corr, lags): n = len(rows) updated = datetime.now().strftime('%Y-%m-%d %H:%M:%S') first_ts = rows[0]['ts'] if rows else '-' last_ts = rows[-1]['ts'] if rows else '-' col_headers = ['因子', '与总分', '同期价格'] + [f'+{lag}步价格' for lag in lags if lag > 0] lag_keys = ['vs_total', 'vs_price_lag0'] + [f'vs_price_lag{lag}' for lag in lags if lag > 0] lag_labels = ['与总分相关', '同期价格相关'] + [f'前瞻{lag}步({lag}分钟)' for lag in lags if lag > 0] rows_html = '' # 按「与总分」相关性排序 sorted_keys = sorted(FACTOR_KEYS, key=lambda k: -abs(corr[k].get('vs_total', 0))) for k in sorted_keys: name = FACTOR_NAMES[k] c = corr[k] cells = f'{k}
{name}' for lk in lag_keys: v = c.get(lk, 0) bg = color_for(v) sign = '+' if v >= 0 else '' cells += f'{sign}{v:.3f}' rows_html += f'{cells}\n' # 统计摘要 strong_pos = [(k, corr[k]['vs_price_lag1']) for k in FACTOR_KEYS if corr[k].get('vs_price_lag1',0) > 0.15] strong_neg = [(k, corr[k]['vs_price_lag1']) for k in FACTOR_KEYS if corr[k].get('vs_price_lag1',0) < -0.15] strong_pos.sort(key=lambda x: -x[1]) strong_neg.sort(key=lambda x: x[1]) insight_html = '' return f""" 因子拟合度分析

📊 因子拟合度分析报告

数据点: {n} 条  |  时间范围: {first_ts[:16]} ~ {last_ts[:16]}  |  更新: {updated}  |  2分钟自动刷新
■ 强正相关(>0.3) ■ 弱正相关 ■ 近无相关 ■ 弱负相关 ■ 强负相关(<-0.3)

🔍 关键发现

{insight_html}
{''.join(f'' for h in col_headers)} {''.join(f'' for l in lag_labels)} {rows_html}
{h}
{l}
说明：
• 与总分相关：该因子与当前综合得分的皮尔逊相关系数，反映因子对总分的贡献一致性
• 同期价格相关：因子当前值与同期归一化价格的相关性（正=同向，负=反向指标）
• 前瞻N步价格：因子当前值与未来N分钟对数收益的相关性，值越高=预测能力越强
• 前瞻相关性正负号：正=因子高→价格涨，负=因子高→价格跌（反向指标）
• 每步约1分钟（取决于交易循环间隔）
""" def print_table(corr, lags): """终端输出""" lag_keys = ['vs_total', 'vs_price_lag0'] + [f'vs_price_lag{lag}' for lag in lags if lag > 0] header = f"{'因子':<6} {'与总分':>8} {'同期价格':>10}" + ''.join(f" {'前瞻'+str(l)+'步':>9}" for l in lags if l > 0) print(header) print('-' * len(header)) sorted_keys = sorted(FACTOR_KEYS, key=lambda k: -abs(corr[k].get('vs_total', 0))) for k in sorted_keys: c = corr[k] row = f"{k+' '+FACTOR_NAMES[k]:<10}" for lk in lag_keys: v = c.get(lk, 0) row += f" {v:>+8.3f}" print(row) def main(): gen_html = '--html' in sys.argv print(f"解析日志 {LOG_FILE}...") rows = parse_log(LOG_FILE, MAX_POINTS) if not rows: print("无数据") return print(f"有效数据点: {len(rows)} 时间范围: {rows[0]['ts'][:16]} ~ {rows[-1]['ts'][:16]}") lags = (0, 1, 3, 5, 10) print("\n计算相关系数...") corr = compute_corr(rows, lags) print(f"\n{'='*65}") print(f"因子拟合度报告 (基于最近{len(rows)}条数据)") print(f"{'='*65}") print_table(corr, lags) print(f"\n{'='*65}") print("关键发现：") # 前瞻+1步正相关（预测性做多因子） sp = [(k, corr[k].get('vs_price_lag1',0)) for k in FACTOR_KEYS if corr[k].get('vs_price_lag1',0) > 0.10] sp.sort(key=lambda x: -x[1]) if sp: print(f" ✅ 前瞻1步正相关（预测涨）: {', '.join(f'{k}({v:+.3f})' for k,v in sp)}") sn = [(k, corr[k].get('vs_price_lag1',0)) for k in FACTOR_KEYS if corr[k].get('vs_price_lag1',0) < -0.10] sn.sort(key=lambda x: x[1]) if sn: print(f" 🔄 前瞻1步负相关（反向因子）: {', '.join(f'{k}({v:+.3f})' for k,v in sn)}") rev = [(k, corr[k].get('vs_price_lag0',0)) for k in FACTOR_KEYS if corr[k].get('vs_price_lag0',0) < -0.10] rev.sort(key=lambda x: x[1]) if rev: print(f" ⚠️ 同期反向指标: {', '.join(f'{k}({v:+.3f})' for k,v in rev)}") weak = [k for k in FACTOR_KEYS if all(abs(corr[k].get(lk,0)) < 0.08 for lk in ['vs_total','vs_price_lag0','vs_price_lag1','vs_price_lag3'])] if weak: print(f" ❌ 低效因子（相关性<0.08）: {', '.join(weak)}") if gen_html: html = build_html_report(rows, corr, lags) with open(OUT_HTML, 'w', encoding='utf-8') as f: f.write(html) print(f"\nHTML报告已生成: {OUT_HTML}") import subprocess subprocess.Popen(['start', OUT_HTML], shell=True) # 保存JSON结果 out = { 'generated': datetime.now().isoformat(), 'n_points': len(rows), 'time_range': [rows[0]['ts'], rows[-1]['ts']], 'correlations': {k: {lk: v for lk, v in c.items()} for k, c in corr.items()}, } out_json = os.path.join(_ROOT, 'corr_results.json') with open(out_json, 'w', encoding='utf-8') as f: json.dump(out, f, ensure_ascii=False, indent=2) print(f"JSON结果已保存: {out_json}") if __name__ == '__main__': main()