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

feat: v_next3生产落地——板块ADX入板块上下文+行业牛杠杆接入position_advice

Browse Source
This commit is contained in:
hmo
2026-07-30 21:59:20 +08:00
parent 82ec79a467
commit d28bf7fc3b
2 changed files with 42 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
strategy_lab.py
+21 -6
View File
@@ -552,17 +552,21 @@ def prepare_sector_context(start_date, end_date):
try:
# 1. 板块指数历史（全周期）
try:
idx_rows = conn.execute("""
SELECT sector, date, close, change_pct FROM sector_index_daily
WHERE date >= ? AND date <= ? ORDER BY sector, date
""", (start_date, end_date)).fetchall()
idx_rows = conn.execute(
"SELECT sector, date, close, change_pct, high, low FROM sector_index_daily "
"WHERE date >= ? AND date <= ? ORDER BY sector, date",
(start_date, end_date)).fetchall()
except sqlite3.OperationalError:
idx_rows = []
# 每板块计算 MA20 和斜率
from collections import defaultdict
by_sector = defaultdict(list)
for sec, d, close, chg in idx_rows:
sec_hl = defaultdict(list)
for row in idx_rows:
sec, d, close, chg = row[0], row[1], row[2], row[3]
by_sector[sec].append((d, close, chg))
if len(row) >= 6:
sec_hl[sec].append((d, row[4], row[5]))
for sec, series in by_sector.items():
closes = [c for _, c, _ in series]
for i, (d, close, chg) in enumerate(series):
@@ -574,8 +578,19 @@ def prepare_sector_context(start_date, end_date):
ma20_5 = sum(closes[i-24:i-4]) / 20
if ma20_5 > 0:
slope = round((ma20 - ma20_5) / ma20_5 * 100, 3)
_adx = None
_hl = sec_hl.get(sec, [])
if len(_hl) >= 20 and i >= 14:
from backtest_framework import calc_trend_strength
_hs = [x[1] for x in _hl]
_ls = [x[2] for x in _hl]
_cs = [c for _, c, _ in series]
if len(_cs) == len(_hl):
_av = calc_trend_strength(_hs, _ls, _cs, 14)
if i < len(_av):
_adx = _av[i]
_SECTOR_CTX.setdefault(d, {})[sec] = {
'change': chg, 'above_ma20': above, 'slope': slope,
'change': chg, 'above_ma20': above, 'slope': slope, 'adx': _adx,
}
# 2. sector_snapshots 补充净流入和涨幅（近期，THS命名）
snap_rows = conn.execute("""