#!/usr/bin/env python3 """dashboard_data.py — 汇总涨停回调战法数据 → dashboard_data.json（看板页面消费） v2：多战法架构（strategy 字段）+ 指数曲线联动 + 全量历史信号/平仓 数据源：strength_baseline.csv + sim_trading_state.json + sim_trading_nav.csv + 指数日线(tushare) """ import json, os, sys, subprocess, re import pandas as pd, numpy as np BASE = '/Users/xpresso/zt_app/backtest_zt_full' QUANT = '/Users/xpresso/zt_app' OUT = '/Users/xpresso/zt_app/dashboard_data.json' PN = {'hs300': '沪深300', 'zz500': '中证500', 'zz1000': '中证1000', 'zz2000': '中证2000'} IDX = [('hs300', '000300.SH'), ('zz500', '000905.SH'), ('zz1000', '000852.SH'), ('zz2000', '932000.CSI')] STRATEGY = 'zt-huitiao' # 当前唯一战法：涨停回调 def fetch_indices(dates_needed): """拉四指数日线（有本地缓存 index_cache.csv），返回 {code: [(date, close)]}""" cache = f'{QUANT}/index_cache.csv' need = set(dates_needed) have = {} if os.path.exists(cache): c = pd.read_csv(cache, dtype={'trade_date': str}) for _, r in c.iterrows(): have.setdefault(r['ts_code'], {})[r['trade_date']] = r['close'] # 缺哪些日期 missing = need - {d for v in have.values() for d in v} else: missing = need if missing: import tushare as ts src = open(f'{QUANT}/code/plot_sim_live.py').read() tok = re.search(r"pro_api\('([^']+)'\)", src) pro = ts.pro_api(tok.group(1)) if tok else ts.pro_api() rows = [] lo = min(missing); hi = max(missing) for name, code in IDX: ix = pro.index_daily(ts_code=code, start_date=lo, end_date=hi) if ix is not None and len(ix): for _, r in ix.iterrows(): rows.append({'ts_code': code, 'trade_date': str(r['trade_date']), 'close': float(r['close'])}) have.setdefault(code, {})[str(r['trade_date'])] = float(r['close']) if rows: # 合并写缓存 newc = pd.DataFrame(rows).drop_duplicates(subset=['ts_code', 'trade_date']) if os.path.exists(cache): old = pd.read_csv(cache, dtype={'trade_date': str}) newc = pd.concat([old, newc]).drop_duplicates(subset=['ts_code', 'trade_date'], keep='last') newc.to_csv(cache, index=False) out = {} for name, code in IDX: d = have.get(code, {}) out[name] = sorted([(dt, v) for dt, v in d.items() if dt in need]) return out def main(): # ---- 模拟盘状态 ---- st = json.load(open(f'{QUANT}/sim_trading_state.json')) nav = pd.read_csv(f'{QUANT}/sim_trading_nav.csv') sb = pd.read_pickle(f'{BASE}/stock_basic.pkl') name_map = dict(zip(sb['ts_code'], sb['name'])) ind_map = dict(zip(sb['ts_code'], sb['industry'])) # ---- baseline ---- bdf = pd.read_csv(f'{BASE}/strength_baseline.csv', dtype={'buy_date': str, 'sell_date': str}) today = str(int(max(float(d) for d in nav['date']))) done = bdf.dropna(subset=['ret']) tiers = {} for label, lo, hi in [('强', 85, 999), ('中', 50, 85), ('弱', -1, 50)]: seg = done[(done['strength'] >= lo) & (done['strength'] < hi)] tiers[label] = {'n': int(len(seg)), 'avg': round(float(seg['ret'].mean()) * 100, 2) if len(seg) else None, 'win': round(float((seg['ret'] > 0).mean()) * 100, 1) if len(seg) else None} # ---- 指数曲线（对齐净值日期轴）---- nav_dates = [str(int(float(d))) for d in nav['date']] try: idx_data = fetch_indices(nav_dates + [today]) except Exception as e: print(f'指数拉取失败(继续无指数): {e}') idx_data = {name: [] for name, _ in IDX} # 归一化到首日=1.0 的基准（对齐模拟盘基准日 20260901） idx_series = {} for name, code in IDX: pts = idx_data.get(name, []) base_date = nav_dates[0] if nav_dates else None arr = dict(pts) if base_date and base_date in arr: b0 = arr[base_date] idx_series[name] = [[d, round(arr[d] / b0, 5)] for d in nav_dates if d in arr] else: idx_series[name] = [[d, round(v / pts[0][1], 5)] for d, v in pts] if pts else [] # ---- 持仓明细 ---- positions = [] for code, p in st['positions'].items(): pool = p['pool'] dfp = f"{BASE}/daily_{pool}/{code}.csv" last_px = None if os.path.exists(dfp): try: d = pd.read_csv(dfp) last = d[d['trade_date'].astype(str) == today] if len(last): last_px = float(last.iloc[0]['close']) elif len(d): last_px = float(d.sort_values('trade_date').iloc[-1]['close']) except Exception: pass if last_px is None: continue flt = last_px / p['buy_px'] - 1 positions.append({ 'code': code, 'name': name_map.get(code, code), 'industry': ind_map.get(code, '?'), 'pool': pool, 'pool_cn': PN.get(pool, pool), 'buy_date': p['buy_date'], 'buy_px': round(p['buy_px'], 3), 'last_px': round(last_px, 2), 'shares': int(p['shares']), 'spent': round(p['spent'], 0), 'value': round(last_px * p['shares'], 0), 'float_pnl': round((last_px * p['shares']) - p['spent'], 0), 'float_pct': round(flt * 100, 2), 'strength': round(float(p['std']), 1), }) positions.sort(key=lambda x: -x['float_pct']) # ---- 平仓（模拟盘，全量）---- closed = [{ 'code': c['code'], 'name': name_map.get(c['code'], c['code']), 'pool_cn': PN.get(c['pool'], c['pool']), 'buy_date': c['buy_date'], 'sell_date': c['sell_date'], 'reason': c.get('reason'), 'ret_net': round(c['ret_net'] * 100, 2), 'pnl': round(c['pnl'], 0), 'std': round(c['std'], 1), 'spent': round(c.get('spent', 0), 0), 'sell_amount': round(c.get('sell_px', 0) * c.get('shares', 0), 0), } for c in st.get('closed', [])] closed.sort(key=lambda x: x['sell_date'], reverse=True) # ---- 近期信号（近3交易日未平仓）---- latest_dates = sorted(bdf['buy_date'].unique())[-3:] recent_sig = bdf[bdf['buy_date'].isin(latest_dates) & bdf['ret'].isna()].copy() sigs = [{ 'code': t['code'], 'name': name_map.get(t['code'], t['code']), 'industry': ind_map.get(t['code'], '?'), 'pool_cn': PN.get(t['pool'], t['pool']), 'buy_date': t['buy_date'], 'strength': round(float(t['strength']), 1), 'f6': round(float(t['f6_tb_strength']) * 100, 1) if pd.notna(t['f6_tb_strength']) else None, 'f3': round(float(t['f3_ta_shrink']), 2) if pd.notna(t['f3_ta_shrink']) else None, 'f8': float(t['f8_chain_len']) if pd.notna(t['f8_chain_len']) else None, 'signal_date': str(int(float(t['buy_date']))) if pd.notna(t['buy_date']) else '', } for _, t in recent_sig.iterrows()] sigs.sort(key=lambda x: -x['strength']) # ---- 全量历史信号（含未平仓）：baseline 全部行 ---- # 1) 找模拟盘实际采纳的信号（code+buy_date 匹配 sim 持仓/平仓） sim_taken = set() for code, p in st.get('positions', {}).items(): sim_taken.add(f"{code}_{p['buy_date']}") for c in st.get('closed', []): sim_taken.add(f"{c['code']}_{c['buy_date']}") # 2) 未完成信号用 SELL_COMBO 规则模拟卖出（截至最新交易日） sys.path.insert(0, '/Users/xpresso/zt_app/code') import grid_or as g from scan_daily_tb import simulate_sell, SELL_COMBO, SELL_CN dailies_cache = {} def get_dailies(pool): if pool not in dailies_cache: _, dl = g.load_pool(pool) dailies_cache[pool] = dl return dailies_cache[pool] allb = bdf.copy() # 补未完成信号：用 simulate_sell 算当前浮盈/假想卖点 hist_sigs = [] open_rows = allb[allb['ret'].isna()] done_rows = allb[allb['ret'].notna()] today_s = today def sig_row(t, taken, simulated=None): """统一行构造。simulated=simulate_sell 结果（未平仓信号的假想卖出）""" code = t['code']; bd = str(t['buy_date']) if pd.notna(t['ret']): ret = round(float(t['ret']) * 100, 2) sell_date = str(int(float(t['sell_date']))) if pd.notna(t['sell_date']) else '' reason = t['reason'] if isinstance(t['reason'], str) else '' hold = float(t['hold_days']) if pd.notna(t['hold_days']) else None status = 'closed' elif simulated: ret = round(simulated['ret_net'] * 100, 2) sell_date = str(simulated['sell_date']) reason = simulated['reason'] hold = float(simulated['hold_days']) status = 'closed_sim' # 按规则已触发卖出（今天卖出） else: # 还在持仓中：算当前浮盈 pool = t['pool'] d = get_dailies(pool).get(code) ret = None; last_px = None if d is not None: rows_ = d[d['trade_date'].astype(str) == today_s] if len(rows_): last_px = float(rows_.iloc[0]['close']) if last_px is None and d is not None and len(d): last_px = float(d.sort_values('trade_date').iloc[-1]['close']) if last_px is not None: rows0 = d[d['trade_date'].astype(str) == bd] if len(rows0): buy_px = float(rows0.iloc[0]['open']) * 1.001 ret = round((last_px / buy_px - 1) * 100, 2) sell_date = ''; reason = ''; hold = None status = 'holding' def fv(fc, rnd=2): try: return round(float(t[fc]), rnd) if pd.notna(t[fc]) else None except Exception: return None if status.startswith('closed') and not reason: reason = 'unknown' return {'code': code, 'name': name_map.get(code, code), 'pool_cn': PN.get(t['pool'], t['pool']), 'buy_date': bd, 'sell_date': sell_date, 'strength': round(float(t['strength']), 1), 'ret_net': ret, 'reason': reason, 'hold_days': hold, 'status': status, 'sim_taken': f"{code}_{bd}" in taken, 'shrink': fv('f3_ta_shrink'), 'chain': fv('f8_chain_len', 0), 'ta_gap': fv('f5_ta_gap', 0), 'ta_dd': fv('f4_ta_drawdown', 3), 'ma_align': fv('f11_tb_ma_align', 0), 'tb_gain': fv('f6_tb_strength', 3)} # 已完成信号（缺 reason/hold_days 的用 simulate_sell 规则反推补齐） for _, t in done_rows.iterrows(): t2 = t if not (isinstance(t.get('reason'), str) and t.get('reason')) or pd.isna(t.get('hold_days')): try: d = get_dailies(t['pool']).get(t['code']) except Exception: d = None if d is not None: try: sim = simulate_sell(d, {'code': t['code'], 'buy_date': str(t['buy_date'])}, SELL_COMBO[t['pool']]) except Exception: sim = None if sim is not None: t2 = t.copy() t2['reason'] = sim['reason'] t2['hold_days'] = sim['hold_days'] # 卖出日期若 baseline 已有则保留（官方口径），否则用模拟的 if pd.isna(t.get('sell_date')) or str(t.get('sell_date')) in ('nan', ''): t2['sell_date'] = sim['sell_date'] hist_sigs.append(sig_row(t2, sim_taken)) # 未完成信号（模拟卖出 or 持仓中） for _, t in open_rows.iterrows(): try: d = get_dailies(t['pool']).get(t['code']) except Exception: d = None simulated = None if d is not None: try: simulated = simulate_sell(d, {'code': t['code'], 'buy_date': str(t['buy_date'])}, SELL_COMBO[t['pool']]) except Exception: simulated = None hist_sigs.append(sig_row(t, sim_taken, simulated)) hist_sigs.sort(key=lambda x: x['buy_date'], reverse=True) # 按买入日倒序 # ---- 月度市场环境(四池指数月度均值%) ---- try: idx_c = pd.read_csv(f'{QUANT}/index_cache.csv', dtype={'trade_date': str}) idx_c['ym'] = idx_c['trade_date'].str[:6] mc = idx_c.groupby(['ts_code', 'ym'])['close'].agg(['first', 'last']) mc['chg'] = mc['last'] / mc['first'] - 1 env_m = (mc.groupby('ym')['chg'].mean() * 100).round(2) except Exception: env_m = pd.Series(dtype=float) # ---- 信号质量月度分布 ---- done_h = [h for h in hist_sigs if h['ret_net'] is not None] months = {} for h in done_h: ym = h['buy_date'][:6] m = months.setdefault(ym, {'strong': 0, 'mid': 0, 'weak': 0, 'strong_ret': [], 'all_ret': []}) tier = 'strong' if h['strength'] >= 85 else ('mid' if h['strength'] >= 50 else 'weak') m[tier] += 1 m['all_ret'].append(h['ret_net']) if tier == 'strong': m['strong_ret'].append(h['ret_net']) monthly = [] for ym in sorted(months): m = months[ym] avg_strong = sum(m['strong_ret']) / len(m['strong_ret']) if m['strong_ret'] else None # 强+中信号胜率 sm = [h for h in done_h if h['buy_date'][:6] == ym and h['strength'] >= 50] win_sm = round(sum(1 for h in sm if h['ret_net'] > 0) / len(sm) * 100, 1) if sm else None monthly.append({'ym': ym[:4] + '-' + ym[4:], 'strong': m['strong'], 'mid': m['mid'], 'weak': m['weak'], 'total': m['strong'] + m['mid'] + m['weak'], 'avg_strong_ret': round(avg_strong, 2) if avg_strong is not None else None, 'win_sm': win_sm, 'env': float(env_m.get(ym, float('nan')))}) # ---- 今日交易（模拟盘当日买卖）---- today_buys = [] today_sells = [] for c in st.get('closed', []): if str(c['sell_date']) == today_s: today_sells.append({'name': name_map.get(c['code'], c['code']), 'code': c['code'], 'ret_net': round(c['ret_net'] * 100, 2), 'pnl': round(c['pnl'], 0), 'reason': c.get('reason'), 'std': round(c['std'], 1), 'pool_cn': PN.get(c['pool'], c['pool']), 'buy_px': round(c.get('buy_px', 0), 2), 'sell_px': round(c.get('sell_px', 0), 2), 'buy_date': c['buy_date'], 'spent': round(c.get('spent', 0), 0), 'sell_amount': round(c.get('sell_px', 0) * c.get('shares', 0), 0)}) for code, p in st.get('positions', {}).items(): if str(p['buy_date']) == today_s: today_buys.append({'name': name_map.get(code, code), 'code': code, 'std': round(float(p['std']), 1), 'buy_px': round(p['buy_px'], 3), 'pool_cn': PN.get(p['pool'], p['pool'])}) today_trades = {'date': today_s, 'buys': today_buys, 'sells': today_sells} # ---- 净值序列（含每日持仓数, 供联动）---- navs = [{'date': str(int(float(r['date']))), 'nav': round(float(r['total']), 0), 'n_pos': int(r['n_pos'])} for _, r in nav.iterrows()] # ---- K线窗口数据提取 ---- def kline_for(code, pool, buy_date, extra=None, w_row=None, to_latest=False): """取 buy_date 前 30 到后 20 个交易日的K线窗口。返回 {dates,o,h,l,c,v,buy_i,sell_i}""" dfp = f"{BASE}/daily_{pool}/{code}.csv" if not os.path.exists(dfp): return None try: d = pd.read_csv(dfp, dtype={'trade_date': str}) d = d.sort_values('trade_date').reset_index(drop=True) bi = d.index[d['trade_date'] == str(buy_date)] if len(bi) == 0: return None bi = int(bi[0]) lo = max(0, bi - 25) hi = len(d) if to_latest else min(len(d), bi + 15) w = d.iloc[lo:hi] sell_i = None if extra and extra.get('sell_date'): si = d.index[d['trade_date'] == str(extra['sell_date'])] if len(si): sell_i = int(si[0]) - lo # t0/ta/tb 定位：tb=突破日=买入日前一交易日; t0=tb-f7_gap; ta=t0+f5_gap t0_i = ta_i = None try: f7 = int(float(w_row['f7_tb_gap'])) if pd.notna(w_row.get('f7_tb_gap')) else None f5 = int(float(w_row['f5_ta_gap'])) if pd.notna(w_row.get('f5_ta_gap')) else None tb_i = bi - lo - 1 if f7 is not None: t0_i = tb_i - f7 if t0_i is not None and f5 is not None: ta_i = t0_i + f5 except Exception: pass def r1(x): return float(round(x, 1)) return {'d': w['trade_date'].tolist(), # 完整 YYYYMMDD 'o': [r1(x) for x in w['open']], 'h': [r1(x) for x in w['high']], 'l': [r1(x) for x in w['low']], 'c': [r1(x) for x in w['close']], 'v': [int(round(x)) for x in w['vol']], 'buy_i': bi - lo, 'sell_i': sell_i, 't0_i': t0_i, 'ta_i': ta_i, 'tb_i': bi - lo - 1} except Exception: return None # ---- 持仓K线墙（模拟盘实际持仓, 尾部延伸到最新交易日）---- pos_klines = [] bdf_code_buy = bdf.set_index(['code', 'buy_date']) if 'buy_date' in bdf.columns else None for p in sorted(st['positions'].items(), key=lambda kv: (-kv[1].get('std', 0))): code, pinfo = p pool = None w_row = None if bdf_code_buy is not None and (code, str(pinfo['buy_date'])) in bdf_code_buy.index: r_ = bdf_code_buy.loc[(code, str(pinfo['buy_date']))] pool = r_['pool'] if pd.notna(r_.get('pool')) else pool w_row = r_ if pool is None: # 退路: 从缓存池找 for pp in (list(pools) if 'pools' in dir() else ['hs300','zz500','zz1000','zz2000']): if os.path.exists(f'{BASE}/daily_{pp}/{code}.csv'): pool = pp; break if pool is None: continue k = kline_for(code, pool, str(pinfo['buy_date']), w_row=w_row, to_latest=True) if k: # 最新收盘(窗口尾) 相对买入价(开盘x1.001) 幅度 try: buypx = k['o'][k['buy_i']] * 1.001 lastpx = k['c'][-1] float_pct = (lastpx / buypx - 1) * 100 except Exception: float_pct = None pos_klines.append({**pinfo, 'code': code, 'pool': pool, 'name': name_map.get(code, code), 'pool_cn': PN.get(pool, pool), 'industry': ind_map.get(code, '?'), 'strength': float(pinfo.get('std', 0)), 'k': k, 'float_pct': round(float_pct, 2) if float_pct is not None else None}) # ---- 顶部强信号K线（近期强度≥85 前8个, 内嵌）---- pool_of = {} for _, t in bdf.iterrows(): pool_of.setdefault(t['code'], t['pool']) strong_recent = [s for s in sigs if s['strength'] >= 85][:9] strong_k = [] for s in strong_recent: pool = pool_of.get(s['code']) k = kline_for(s['code'], pool, s['buy_date'], w_row=recent_sig[recent_sig['code']==s['code']].iloc[0] if len(recent_sig[recent_sig['code']==s['code']]) else None) if pool else None if k: strong_k.append({**s, 'pool': pool, 'k': k}) # ---- 历史信号K线缓存（按需文件; 强档预生成2000笔）---- KC_DIR = f'{QUANT}/kline_cache' os.makedirs(KC_DIR, exist_ok=True) hist_kline_index = {} # 全量信号K线缓存（含全部历史+未平仓；窗口已缩至40根, 体积可控） open_for_k = allb[allb['ret'].isna()] def cache_kline(t, sell_date=None): fn = f"{t['code'].replace('.','_')}_{t['buy_date']}.json" path = f"{KC_DIR}/{fn}" sd = None if sell_date is not None and pd.notna(sell_date): sd = str(int(float(sell_date))) # float 20240301.0 → '20240301' if not os.path.exists(path): k = kline_for(t['code'], t['pool'], t['buy_date'], {'sell_date': sd}, w_row=t) if k is None: return None json.dump(k, open(path, 'w')) hist_kline_index[f"{t['code']}_{t['buy_date']}"] = fn return fn for _, t in done_rows.iterrows(): cache_kline(t, t.get('sell_date')) for _, t in open_for_k.iterrows(): cache_kline(t, None) # ---- 滚动30日窗口(point-in-time): 熔断与对照口径 ---- import bisect tdays_all = sorted(bdf['buy_date'].unique()) sig_recs = [] for _, t in allb.iterrows(): fp = f"{BASE}/daily_{t['pool']}/{t['code']}.csv" if not os.path.exists(fp): continue d = pd.read_csv(fp, dtype={'trade_date': str}).sort_values('trade_date') row = d[d['trade_date'] == str(t['buy_date'])] if not len(row): continue sig_recs.append({'b': str(t['buy_date']), 'st': float(t['strength']), 'sd': str(int(float(t['sell_date']))) if pd.notna(t['sell_date']) else None, 'ret': float(t['ret']) if pd.notna(t['ret']) else None, 'dates': d['trade_date'].tolist(), 'closes': d['close'].astype(float).values, 'bp': float(row.iloc[0]['open']) * 1.001}) def close_at2(r, dd): i = bisect.bisect_right(r['dates'], dd) - 1 return r['closes'][i] if i >= 0 else None # 沪深300收盘索引(供窗口对照) hs_c = {} try: idx_c2 = pd.read_csv(f'{QUANT}/index_cache.csv', dtype={'trade_date': str}) hs_df = idx_c2[idx_c2['ts_code'] == '000300.SH'].sort_values('trade_date') hs_c = dict(zip(hs_df['trade_date'].tolist(), hs_df['close'].astype(float).values)) except Exception: hs_c = {} hs_dates = sorted(hs_c.keys()) def hs_at(dd): # dd当日或之前最近收盘 if not hs_c: return None i = bisect.bisect_right(hs_dates, dd) - 1 return hs_c[hs_dates[i]] if i >= 0 else None roll30 = [] n_td = len(tdays_all) for di, dd in enumerate(tdays_all): # 30个交易日窗口: 需前面有29个交易日, 否则是残缺窗口→跳过(不纳入序列/分位) if di < 29: continue t_lo = tdays_all[di - 29] sm_r, st_r, n_in = [], [], 0 for r in sig_recs: if r['b'] < t_lo or r['b'] > dd: continue n_in += 1 if r['ret'] is not None and r['sd'] and r['sd'] <= dd: v = r['ret'] else: c = close_at2(r, dd) if c is None: continue v = c / r['bp'] - 1 if r['st'] >= 50: sm_r.append(v) if r['st'] >= 85: st_r.append(v) if n_in >= 10 and sm_r: p0 = hs_at(t_lo); p1 = hs_at(dd) hs30 = round((p1 / p0 - 1) * 100, 2) if (p0 and p1 and p0 > 0) else None roll30.append({'date': dd, 'n': n_in, 'n_sm': len(sm_r), 'n_strong': len(st_r), 'win_sm': round(sum(1 for x in sm_r if x > 0) / len(sm_r) * 100, 1), 'win_strong': round(sum(1 for x in st_r if x > 0) / len(st_r) * 100, 1) if st_r else None, 'ret_sm': round(float(np.mean(sm_r)) * 100, 2), 'ret_strong': round(float(np.mean(st_r)) * 100, 2) if st_r else None, 'hs30': hs30}) # ---- 按卖出日归组的结算口径(roll30s): 窗口内完成卖出的交易, 全部真实净收益 ---- roll30s = [] for di, dd in enumerate(tdays_all): if di < 29: continue t_lo = tdays_all[di - 29] sm_r2, st_r2, n_s2 = [], [], 0 for r in sig_recs: # 卖出日在窗口内 [t_lo, dd] 才纳入 if not (r['sd'] and t_lo <= r['sd'] <= dd): continue v = r['ret'] # 已平仓: 实际净收益(含全成本) if v is None: continue n_s2 += 1 if r['st'] >= 50: sm_r2.append(v) if r['st'] >= 85: st_r2.append(v) if n_s2 >= 5 and sm_r2: p0 = hs_at(t_lo); p1 = hs_at(dd) hs30_2 = round((p1 / p0 - 1) * 100, 2) if (p0 and p1 and p0 > 0) else None roll30s.append({'date': dd, 'n': n_s2, 'n_sm': len(sm_r2), 'n_strong': len(st_r2), 'win_sm': round(sum(1 for x in sm_r2 if x > 0) / len(sm_r2) * 100, 1), 'win_strong': round(sum(1 for x in st_r2 if x > 0) / len(st_r2) * 100, 1) if st_r2 else None, 'ret_sm': round(float(np.mean(sm_r2)) * 100, 2), 'ret_strong': round(float(np.mean(st_r2)) * 100, 2) if st_r2 else None, 'hs30': hs30_2}) roll30s_last = roll30s[-1] if roll30s else None roll30s_pct = None if roll30s_last and len(roll30s) > 20: wins2 = [r['win_sm'] for r in roll30s[:-1]] roll30s_pct = {'win': round(sum(1 for x in wins2 if x < roll30s_last['win_sm']) / len(wins2) * 100, 0)} # 历史分位(当前窗口 vs 全历史) cur = roll30[-1] if roll30 else None roll_pct = None if cur and len(roll30) > 20: wins = [r['win_sm'] for r in roll30[:-1]] rets = [r['ret_sm'] for r in roll30[:-1]] roll_pct = {'win': round(sum(1 for x in wins if x < cur['win_sm']) / len(wins) * 100, 0), 'ret': round(sum(1 for x in rets if x < cur['ret_sm']) / len(rets) * 100, 0)} out = { 'strategy': STRATEGY, 'strategy_cn': '涨停回调', 'generated_at': today, 'sim': { 'cash': round(st['cash'], 0), 'n_pos': len(st['positions']), 'n_closed': len(closed), 'realized': round(sum(c['pnl'] for c in closed), 0), 'floating': round(sum(x['float_pnl'] for x in positions), 0), 'total': round(float(nav.iloc[-1]['total']), 0), 'init': 100000, 'params': st.get('params', {}), }, 'nav': navs, 'index_series': idx_series, 'positions': positions, 'closed': closed, 'signals_recent': sigs, 'signals_history': hist_sigs, 'monthly': monthly, 'roll30': roll30, 'roll30_last': cur, 'roll30_pct': roll_pct, 'roll30s': roll30s, 'roll30s_last': roll30s_last, 'roll30s_pct': roll30s_pct, 'today_trades': today_trades, 'strong_klines': strong_k, 'pos_klines': pos_klines, 'hist_kline_index': hist_kline_index, 'tiers': tiers, 'baseline_n': int(len(bdf)), } json.dump(out, open(OUT, 'w'), ensure_ascii=False) print(f'dashboard_data.json 已生成: 持仓{len(positions)} 平仓{len(closed)} 近期信号{len(sigs)} ' f'历史信号{len(hist_sigs)} 指数曲线{ {k: len(v) for k, v in idx_series.items()} } 总资产{out["sim"]["total"]}') if __name__ == '__main__': main()