DockerSpace / models /factor_skill_registry.py
DennisChan0909's picture
Backup current stock predictor strategies
ee37d63
Raw
History Blame Contribute Delete
2.89 kB
"""Progressive-disclosure registry for the audited 44-factor overlay."""
from __future__ import annotations
from typing import Any
from models.factor_strategy_catalog import FACTOR_STRATEGIES
GROUP_SOP = {
"oldwang": "docs/factor_skills/group_sops.md#老王趨勢與守線",
"volume": "docs/factor_skills/group_sops.md#量價行為",
"candle": "docs/factor_skills/group_sops.md#k-線與影線",
"institutional": "docs/factor_skills/group_sops.md#法人籌碼",
"chip": "docs/factor_skills/group_sops.md#籌碼與風險",
"technical": "docs/factor_skills/group_sops.md#技術指標",
"momentum": "docs/factor_skills/group_sops.md#中長線動能",
"pattern": "docs/factor_skills/group_sops.md#組合型交易策略",
"regime": "docs/factor_skills/group_sops.md#市場環境",
}
GROUP_TRIGGER = {
"oldwang": "趨勢、均線守線或爆量高低點出現",
"volume": "量比、爆大量、量縮或 60K 爆量低點出現",
"candle": "影線、高基期、十字線或多空交戰出現",
"institutional": "外資、投信、自營商或法人合計資料可用",
"chip": "大戶成交、籌碼集中、融資融券、借券或高波動風險出現",
"technical": "需要 MACD、RSI、布林或均線確認",
"momentum": "需要 20/60/120 日動能分層",
"pattern": "K 線組合型態被觸發",
"regime": "需要盤整、市場或總體風險過濾",
}
def level1_catalog(factor_names: list[str]) -> list[dict[str, Any]]:
"""Return names and descriptions only; do not load full SOP text."""
rows = []
for name in factor_names:
strategy = FACTOR_STRATEGIES[name]
group = strategy["group"]
rows.append(
{
"factor": name,
"group": group,
"group_zh": strategy["group_zh"],
"zh_name": strategy["zh_name"],
"description": strategy["strategy"],
"trigger": GROUP_TRIGGER[group],
"level2_sop": GROUP_SOP[group],
}
)
return rows
def level2_sops_for_active_factors(active_factors: list[str]) -> list[str]:
"""Resolve only the SOP sections needed for triggered factors."""
paths = []
for item in level1_catalog(active_factors):
if item["level2_sop"] not in paths:
paths.append(item["level2_sop"])
return paths
def level3_scripts_for_active_factors(active_factors: list[str]) -> list[str]:
"""Resolve deterministic deep-validation scripts only when needed."""
scripts = []
if {"intraday_60k_volume_low_guard", "intraday_60k_volume_low_break"} & set(active_factors):
scripts.append("scripts/run_44factor_buy_gate_evaluation.py")
if {"big_player_buy", "big_player_sell"} & set(active_factors):
scripts.append("data/realtime_chip_adapter.py")
return scripts