hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: 温区数据按周期计算加缓存(键=period_tag+数据版本)——现场归因30秒/请求,缓存后毫秒级

Browse Source
This commit is contained in:
xxm
2026-08-15 20:53:51 +08:00
parent 2ce88ae436
commit 0b08e5f6cc
1 changed files with 138 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.py
+138 -115
View File
@@ -39,6 +39,142 @@ def _chk_tcp(host, port, timeout=3):
except Exception:
return False
# ── 2026-08-15 温区数据按周期计算 + 缓存（现场温区归因约30秒/请求，缓存后毫秒级）──
# 键 = period_tag + strategy_research 最新 created_at（数据更新即失效）
_regime_winrates_cache = {"key": None, "pt": None, "data": None}
def _compute_regime_winrates_cached(pt, _approx_univ):
"""按 period_tag 从 strategy_research 对应周期 trades 现场温区归因（带缓存）"""
import sqlite3 as _sq
import math as _mth
from collections import defaultdict as _dd
from flask import request as _req
# 缓存键
_cache_key = None
try:
_cc = _sq.connect(str(DATA_DIR / "mofin.db"), timeout=5)
_ver_s = _cc.execute("SELECT MAX(created_at) FROM strategy_research").fetchone()[0]
_cc.close()
_cache_key = (pt or "2y") + "|" + str(_ver_s)
except Exception:
_cache_key = None
if _cache_key and _regime_winrates_cache.get("key") == _cache_key \
and _regime_winrates_cache.get("pt") == (pt or "2y"):
return _regime_winrates_cache.get("data") or {}
_pt_chain = {'1m': '1y', '6m': '1y', '1y': '1y', '2y': '2y', '5y': '5y', '10y': '10y'}
_pt_use = _pt_chain.get(pt or '2y', '2y')
_regime_winrates = {}
try:
_c = _sq.connect(str(DATA_DIR / "mofin.db"), timeout=10)
_c.execute("PRAGMA busy_timeout=10000")
_row_map = {}
for _r in _c.execute(
"SELECT version, market, period_tag, results_json FROM strategy_research "
"ORDER BY version, market, "
"CASE period_tag WHEN '1y' THEN 1 WHEN '2y' THEN 2 WHEN '5y' THEN 3 WHEN '10y' THEN 4 ELSE 5 END"
).fetchall():
_k = (_r[0], _r[1])
if _k not in _row_map:
_row_map[_k] = {}
_row_map[_k][_r[2]] = _r[3]
_rmap_a = dict(_c.execute("SELECT date, regime FROM market_regime WHERE market='a'").fetchall())
_rmap_hk = dict(_c.execute("SELECT date, regime FROM market_regime WHERE market='hk'").fetchall())
_c.close()
def _calc_extra(_trades):
if not _trades:
return {}
_profits = [t.get("profit_pct", 0) for t in _trades]
_wins = [p for p in _profits if p > 0]
_losses = [p for p in _profits if p <= 0]
_wr = len(_wins) / len(_profits) * 100 if _profits else 0
_avg = sum(_profits) / len(_profits) if _profits else 0
_avg_w = sum(_wins) / len(_wins) if _wins else 0
_avg_l = abs(sum(_losses) / len(_losses)) if _losses else 1
_pf = _avg_w / _avg_l if _avg_l > 0 else 0
_mr = _avg / 100
_std = _mth.sqrt(sum((p / 100 - _mr) ** 2 for p in _profits) / (len(_profits) - 1)) if len(_profits) > 1 else 0
_sh = _mr / _std * _mth.sqrt(252) if _std > 0 else 0
_holds = [t.get("hold_days", 0) for t in _trades if t.get("hold_days")]
_ah = sum(_holds) / len(_holds) if _holds else 0
return {"win_rate": round(_wr, 1), "avg_pnl": round(_avg, 2),
"avg_hold_days": round(_ah, 1), "sharpe_ratio": round(_sh, 2),
"profit_factor": round(_pf, 2)}
def _portfolio_sim(_trades, _cap=1000000, _slots=10):
if not _trades:
return {}
try:
from strategy_lab import portfolio_sim
return portfolio_sim(_trades, capital=_cap, max_positions=_slots, cost=True)
except Exception:
return {}
for (_ver, _mkt), _periods in _row_map.items():
_js = None
_cand = None
for _p in [_pt_use, '5y', '10y']:
if _p in _periods and _periods[_p]:
_cand = _periods[_p]
break
if not _cand:
continue
try:
_trades = json.loads(_cand).get("trades", [])
except Exception:
continue
if not _trades:
continue
_rmap = _rmap_a if _mkt != 'hk' else _rmap_hk
_by_regime = _dd(list)
for _t in _trades:
_ed = _t.get("entry_date", "")
if _ed in _rmap:
_by_regime[_rmap[_ed]].append(_t)
for _reg, _reg_trades in _by_regime.items():
if len(_reg_trades) < 2:
continue
_extra = _calc_extra(_reg_trades)
_sim = _portfolio_sim(_reg_trades)
if not _sim:
continue
_wr_v = _extra.get("win_rate")
_cagr_v = _sim.get("cagr_pct")
_ret_v = _sim.get("total_return_pct")
_dd_v = _sim.get("portfolio_max_dd_pct")
_cf_v = _sim.get("capital_final")
_pt_v = _sim.get("positions_taken")
_sh_v = _extra.get("sharpe_ratio")
_pf_v = _extra.get("profit_factor")
_regime_winrates.setdefault(_ver, {})[_reg] = {
"trades": len(_reg_trades),
"win_rate": _wr_v, "avg_pnl": _extra.get("avg_pnl"),
"avg_hold_days": _extra.get("avg_hold_days"),
"total_return_pct": _ret_v, "cagr_pct": _cagr_v,
"max_dd_pct": _dd_v, "capital_final": _cf_v,
"positions_taken": _pt_v, "sharpe_ratio": _sh_v,
"profit_factor": _pf_v,
"period_tag": _pt_use,
"portfolio": {"cagr_pct": _cagr_v, "total_return_pct": _ret_v,
"portfolio_max_dd_pct": _dd_v, "capital_final": _cf_v,
"positions_taken": _pt_v, "sharpe_ratio": _sh_v,
"profit_factor": _pf_v},
"universality": _approx_univ(_ver, _reg, len(_reg_trades)),
}
except Exception:
pass
if _cache_key:
try:
_regime_winrates_cache["key"] = _cache_key
_regime_winrates_cache["pt"] = pt or "2y"
_regime_winrates_cache["data"] = _regime_winrates
except Exception:
pass
return _regime_winrates
def _chk_http(host, port, path, timeout=3):
try:
@@ -565,122 +701,9 @@ def api_research_strategies():
_active_set = set(_weights.get("active") or [])
_hk_market = (_weights.get("markets") or {}).get("hk") or {}
_active_set |= set(_hk_market.get("active") or [])
# ── 2026-08-15 温区数据跟随所选周期（period_tag）──
# 病状：旧版读 strategy_regime_perf 全量表（10y/5y 最长窗口），与前端周期下拉无关，
# 导致选"近2年"表内仍是全量数据（老莫抓包发现，严重误导）。
# 修复：按 period_tag 从 strategy_research 对应周期 trades 现场温区归因（calc_extra+portfolio_sim），
# 周期记录缺失时回退到最近更长周期（1m/6m→1y，2y→2y，5y→5y，10y→10y）。
_regime_winrates = {}
try:
import sqlite3 as _sq
import math as _mth
from collections import defaultdict as _dd
_c = _sq.connect(str(DATA_DIR / "mofin.db"), timeout=10)
_c.execute("PRAGMA busy_timeout=10000")
# 周期回退链：前端可选 1m/6m/1y/2y/5y/10y；strategy_research 有 1y/2y/5y/10y 记录
_pt_chain = {'1m': '1y', '6m': '1y', '1y': '1y', '2y': '2y', '5y': '5y', '10y': '10y'}
_pt_use = _pt_chain.get(pt or '2y', '2y')
# 各策略按周期取 trades（含市场维度；港股策略 market='hk'）
_row_map = {}
for _r in _c.execute(
"SELECT version, market, period_tag, results_json FROM strategy_research "
"ORDER BY version, market, "
"CASE period_tag WHEN '1y' THEN 1 WHEN '2y' THEN 2 WHEN '5y' THEN 3 WHEN '10y' THEN 4 ELSE 5 END"
).fetchall():
_k = (_r[0], _r[1])
if _k not in _row_map:
_row_map[_k] = {}
_row_map[_k][_r[2]] = _r[3]
# 温区映射（A股+港股分开）
_rmap_a = dict(_c.execute("SELECT date, regime FROM market_regime WHERE market='a'").fetchall())
_rmap_hk = dict(_c.execute("SELECT date, regime FROM market_regime WHERE market='hk'").fetchall())
_c.close()
# ── 2026-08-15 温区数据按所选周期计算（函数内缓存，替代全量表/内联重算）──
_regime_winrates = _compute_regime_winrates_cached(pt, _approx_regime_universality)
def _calc_extra(_trades):
if not _trades:
return {}
_profits = [t.get("profit_pct", 0) for t in _trades]
_wins = [p for p in _profits if p > 0]
_losses = [p for p in _profits if p <= 0]
_wr = len(_wins) / len(_profits) * 100 if _profits else 0
_avg = sum(_profits) / len(_profits) if _profits else 0
_avg_w = sum(_wins) / len(_wins) if _wins else 0
_avg_l = abs(sum(_losses) / len(_losses)) if _losses else 1
_pf = _avg_w / _avg_l if _avg_l > 0 else 0
_mr = _avg / 100
_std = _mth.sqrt(sum((p / 100 - _mr) ** 2 for p in _profits) / (len(_profits) - 1)) if len(_profits) > 1 else 0
_sh = _mr / _std * _mth.sqrt(252) if _std > 0 else 0
_holds = [t.get("hold_days", 0) for t in _trades if t.get("hold_days")]
_ah = sum(_holds) / len(_holds) if _holds else 0
return {"win_rate": round(_wr, 1), "avg_pnl": round(_avg, 2),
"avg_hold_days": round(_ah, 1), "sharpe_ratio": round(_sh, 2),
"profit_factor": round(_pf, 2)}
def _portfolio_sim(_trades, _cap=1000000, _slots=10):
if not _trades:
return {}
try:
from strategy_lab import portfolio_sim
return portfolio_sim(_trades, capital=_cap, max_positions=_slots, cost=True)
except Exception:
return {}
for (_ver, _mkt), _periods in _row_map.items():
# 取目标周期 trades（缺失则用最近更长周期）
_js = None
_cand = None
for _p in [_pt_use, '5y', '10y']:
if _p in _periods and _periods[_p]:
_cand = _periods[_p]
break
if not _cand:
continue
try:
_trades = json.loads(_cand).get("trades", [])
except Exception:
continue
if not _trades:
continue
_rmap = _rmap_a if _mkt != 'hk' else _rmap_hk
_by_regime = _dd(list)
for _t in _trades:
_ed = _t.get("entry_date", "")
if _ed in _rmap:
_by_regime[_rmap[_ed]].append(_t)
for _reg, _reg_trades in _by_regime.items():
if len(_reg_trades) < 2:
continue
_extra = _calc_extra(_reg_trades)
_sim = _portfolio_sim(_reg_trades)
if not _sim:
continue
_wr_v = _extra.get("win_rate")
_cagr_v = _sim.get("cagr_pct")
_ret_v = _sim.get("total_return_pct")
_dd_v = _sim.get("portfolio_max_dd_pct")
_cf_v = _sim.get("capital_final")
_pt_v = _sim.get("positions_taken")
_sh_v = _extra.get("sharpe_ratio")
_pf_v = _extra.get("profit_factor")
_regime_winrates.setdefault(_ver, {})[_reg] = {
"trades": len(_reg_trades),
"win_rate": _wr_v, "avg_pnl": _extra.get("avg_pnl"),
"avg_hold_days": _extra.get("avg_hold_days"),
"total_return_pct": _ret_v, "cagr_pct": _cagr_v,
"max_dd_pct": _dd_v, "capital_final": _cf_v,
"positions_taken": _pt_v, "sharpe_ratio": _sh_v,
"profit_factor": _pf_v,
"period_tag": _pt_use,
# 温区级组合级指标（温区行也要显示组合级列）
"portfolio": {"cagr_pct": _cagr_v, "total_return_pct": _ret_v,
"portfolio_max_dd_pct": _dd_v, "capital_final": _cf_v,
"positions_taken": _pt_v, "sharpe_ratio": _sh_v,
"profit_factor": _pf_v},
# 温区级 universality（近似）
"universality": _approx_regime_universality(_ver, _reg, len(_reg_trades)),
}
except Exception:
pass
# 温区级普适前先建立日期→温区映射（A股 market_regime）
_regime_map = {}
try: