89 lines
3.6 KiB
Python
89 lines
3.6 KiB
Python
"""最终挑选校验：对每个板块的候选做"策略实际窗口（最近 120 根）内是否有价格断层"检查。
|
||
|
||
接口的 OHLC 不做拆分/份额折算复原（``pre_close`` 已折算、价格未折算），
|
||
若断层落在最近 120 根内，MA60/ATR 会被污染，网格锚点会算错 → 必须避开。
|
||
|
||
用法： py -3.14 -B labs/analysis/etf/verify_picks.py
|
||
"""
|
||
|
||
import json
|
||
import sys
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(HERE))
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
from backtest import CACHE, DAILY_URL, WARMUP # noqa: E402
|
||
|
||
# 板块 -> (首选, [替补...])
|
||
PLAN = {
|
||
"AI": ("159819.SZ", ["515070.SH", "159363.SZ"]),
|
||
"CPO/通信": ("515880.SH", ["515050.SH", "159994.SZ", "159583.SZ", "159695.SZ"]),
|
||
"PCB": ("159732.SZ", ["562950.SH", "159997.SZ", "561100.SH"]),
|
||
"人形机器人": ("562500.SH", ["159530.SZ", "159770.SZ", "159272.SZ"]),
|
||
"光/光伏": ("515790.SH", ["159755.SZ", "516160.SH", "561910.SH"]),
|
||
"创新药": ("159992.SZ", ["159570.SZ", "515120.SH", "512010.SH"]),
|
||
"半导体": ("588200.SH", ["159516.SZ", "588170.SH", "159995.SZ", "512480.SH"]),
|
||
"存储": ("588750.SH", ["588290.SH", "589130.SH", "588890.SH"]),
|
||
"半导体材料": ("588160.SH", ["588010.SH", "589510.SH", "159761.SZ"]),
|
||
"玻璃基板": ("159745.SZ", ["588010.SH", "159763.SZ"]),
|
||
"电力": ("159326.SZ", ["159611.SZ", "159625.SZ", "561560.SH"]),
|
||
"航空航天": ("159227.SZ", ["512660.SH", "159267.SZ", "512710.SH"]),
|
||
"金属": ("518880.SH", ["512400.SH", "159934.SZ"]),
|
||
"金融": ("512880.SH", ["512070.SH", "512800.SH", "159841.SZ"]),
|
||
"能源": ("515220.SH", ["561360.SH", "159518.SZ"]),
|
||
"宽基(保留)": ("510300.SH", []),
|
||
}
|
||
|
||
|
||
def load(code: str) -> list[dict]:
|
||
path = CACHE / f"{code}.json"
|
||
if path.exists():
|
||
rows = json.loads(path.read_text(encoding="utf-8"))
|
||
else:
|
||
request = urllib.request.Request(f"{DAILY_URL}?code={code}",
|
||
headers={"User-Agent": "Mozilla/5.0"})
|
||
with urllib.request.urlopen(request, timeout=30) as response:
|
||
rows = json.load(response)
|
||
CACHE.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(json.dumps(rows), encoding="utf-8")
|
||
return sorted(rows, key=lambda r: str(r["trade_date"]))
|
||
|
||
|
||
def report(code: str) -> tuple[bool, str]:
|
||
"""返回 (窗口内是否干净, 说明)。窗口 = 最近 120 根（策略指标只用这段）。"""
|
||
rows = load(code)
|
||
window = rows[-120:] if len(rows) >= 120 else rows
|
||
worst, worst_day = 0.0, ""
|
||
for prev, cur in zip(window, window[1:]):
|
||
a, b = float(prev["close"]), float(cur["close"])
|
||
if a > 0:
|
||
change = abs(b / a - 1) * 100
|
||
if change > worst:
|
||
worst, worst_day = change, str(cur["trade_date"])
|
||
clean = worst <= 15
|
||
note = "干净" if clean else f"窗口内{worst_day}跳空 {worst:.1f}%"
|
||
closes = [float(r["close"]) for r in window]
|
||
return clean, f"{note} 最近120根 {min(closes):.3f}~{max(closes):.3f} 最新 {closes[-1]:.3f}"
|
||
|
||
|
||
print(f"{'板块':12} {'首选':11} {'窗口干净':>8} 说明")
|
||
picks = {}
|
||
for sector, (first, alts) in PLAN.items():
|
||
for code in [first, *alts]:
|
||
clean, note = report(code)
|
||
tag = "★首选" if code == first else " 替补"
|
||
print(f"{sector:12} {code:11} {str(clean):>8} {tag} {note}")
|
||
if code == first:
|
||
picks[sector] = (code, clean)
|
||
print()
|
||
print("== 首选中有断层的板块 ==")
|
||
for sector, (code, clean) in picks.items():
|
||
if not clean:
|
||
print(f" {sector}: {code}")
|