feat(quant-engine): 10개 고전 기술전략 갭분석 후 7개 보조신호 채택
사용자 제시 10개 고전 기술전략(골든크로스/모멘텀/52주신고가/연속상승하락/이격도/돌파실패/ 강한종가/변동성확장/평균회귀/추세필터)을 기존 엔진과 대조한 갭분석 결과: - 이미 구현됨: 모멘텀(VELOCITY_V1/RS_MOMENTUM_V1), 이격도·평균회귀(MEAN_REVERSION_GATE_V1) - 신규 채택 7개: GOLDEN_CROSS_SIGNAL_V1, STRONG_CLOSE_SIGNAL_V1, VOLATILITY_EXPANSION_BREAKOUT_V1, FIFTY_TWO_WEEK_HIGH_TRIGGER_V1, CONSECUTIVE_STREAK_V1, BREAKOUT_FAILURE_STOP_V1, TREND_FILTER_GATE_V1 AGENTS.md 하드룰(추격매수 방지, anti-late-entry gate 필수통과)에 따라 BUY 방향 신호 전부를 STRATEGY_SCORING의 보조신호로만 편입 — BREAKOUT_QUALITY_GATE_V2/FOLLOW_THROUGH_DAY_CONFIRM_V1/ ANTI_LATE_ENTRY_GATE_V2 게이트 체인을 우회하는 독립 BUY 트리거로는 사용하지 않음. 검증: validate_specs/validate_golden_coverage_100(100%)/validate_calibration_registry_v1/ validate_schema_model_generation_v1/validate_agents_shrink_v1 전부 PASS. golden test 22/22 PASS. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
#!/usr/bin/env python3
|
||||
"""BREAKOUT_FAILURE_STOP_V1 — spec/formulas/domains/exit.yaml.
|
||||
|
||||
governance/todo/technical_signals_p4_adoption_plan.yaml P4-6.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_OUT = ROOT / "Temp" / "breakout_failure_stop_v1.json"
|
||||
|
||||
|
||||
def evaluate(prior_high: float | None, close_price: float | None, days_since_breakout: int | None) -> dict:
|
||||
if prior_high is None or close_price is None or days_since_breakout is None:
|
||||
return {"breakout_failure": None}
|
||||
breakout_failure = days_since_breakout <= 7 and close_price < prior_high
|
||||
result = {"breakout_failure": breakout_failure}
|
||||
if breakout_failure:
|
||||
result["gate"] = "SELL_RISK_EXIT_REVIEW"
|
||||
result["reason_code"] = "breakout_failure_stop"
|
||||
return result
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--prior-high", type=float, default=None)
|
||||
ap.add_argument("--close", type=float, default=None)
|
||||
ap.add_argument("--days-since-breakout", type=int, default=None)
|
||||
ap.add_argument("--out", default=str(DEFAULT_OUT))
|
||||
args = ap.parse_args()
|
||||
|
||||
result = {
|
||||
"formula_id": "BREAKOUT_FAILURE_STOP_V1",
|
||||
**evaluate(args.prior_high, args.close, args.days_since_breakout),
|
||||
}
|
||||
out = Path(args.out)
|
||||
out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,53 @@
|
||||
#!/usr/bin/env python3
|
||||
"""CONSECUTIVE_STREAK_V1 — spec/formulas/domains/entry.yaml.
|
||||
|
||||
Symmetric up_streak/down_streak from the most recent close-to-close changes.
|
||||
governance/todo/technical_signals_p4_adoption_plan.yaml P4-5.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_OUT = ROOT / "Temp" / "consecutive_streak_v1.json"
|
||||
|
||||
|
||||
def compute_streaks(daily_close_changes: list[float] | None) -> dict:
|
||||
if not daily_close_changes:
|
||||
return {"up_streak": None, "down_streak": None}
|
||||
|
||||
up_streak = 0
|
||||
for change in reversed(daily_close_changes):
|
||||
if change > 0:
|
||||
up_streak += 1
|
||||
else:
|
||||
break
|
||||
|
||||
down_streak = 0
|
||||
for change in reversed(daily_close_changes):
|
||||
if change < 0:
|
||||
down_streak += 1
|
||||
else:
|
||||
break
|
||||
|
||||
return {"up_streak": up_streak, "down_streak": down_streak}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--changes", default=None, help="comma-separated daily close changes")
|
||||
ap.add_argument("--out", default=str(DEFAULT_OUT))
|
||||
args = ap.parse_args()
|
||||
|
||||
daily_close_changes = [float(x) for x in args.changes.split(",")] if args.changes else None
|
||||
result = {"formula_id": "CONSECUTIVE_STREAK_V1", **compute_streaks(daily_close_changes)}
|
||||
out = Path(args.out)
|
||||
out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,42 @@
|
||||
#!/usr/bin/env python3
|
||||
"""FIFTY_TWO_WEEK_HIGH_TRIGGER_V1 — spec/formulas/domains/entry.yaml.
|
||||
|
||||
Feeds BREAKOUT_QUALITY_GATE_V2 -- never an independent BUY trigger.
|
||||
governance/todo/technical_signals_p4_adoption_plan.yaml P4-4.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_OUT = ROOT / "Temp" / "fifty_two_week_high_trigger_v1.json"
|
||||
|
||||
|
||||
def evaluate(close_price: float | None, high52w: float | None) -> bool | None:
|
||||
if close_price is None or high52w is None:
|
||||
return None
|
||||
return close_price >= high52w
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--close", type=float, default=None)
|
||||
ap.add_argument("--high52w", type=float, default=None)
|
||||
ap.add_argument("--out", default=str(DEFAULT_OUT))
|
||||
args = ap.parse_args()
|
||||
|
||||
result = {
|
||||
"formula_id": "FIFTY_TWO_WEEK_HIGH_TRIGGER_V1",
|
||||
"fifty_two_week_high_breakout": evaluate(args.close, args.high52w),
|
||||
"hard_constraint": "feeds_BREAKOUT_QUALITY_GATE_V2_only_not_buy_trigger",
|
||||
}
|
||||
out = Path(args.out)
|
||||
out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,44 @@
|
||||
#!/usr/bin/env python3
|
||||
"""GOLDEN_CROSS_SIGNAL_V1 — spec/formulas/domains/entry.yaml.
|
||||
|
||||
Detects ma20 crossing above ma60 (golden cross). Auxiliary STRATEGY_SCORING signal
|
||||
only -- never an independent BUY trigger. governance/todo/technical_signals_p4_adoption_plan.yaml P4-1.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_OUT = ROOT / "Temp" / "golden_cross_signal_v1.json"
|
||||
|
||||
|
||||
def golden_cross_today(ma20: float | None, ma20_prev: float | None, ma60: float | None, ma60_prev: float | None) -> bool | None:
|
||||
if None in (ma20, ma20_prev, ma60, ma60_prev):
|
||||
return None
|
||||
return ma20_prev <= ma60_prev and ma20 > ma60
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--ma20", type=float, default=None)
|
||||
ap.add_argument("--ma20-prev", type=float, default=None)
|
||||
ap.add_argument("--ma60", type=float, default=None)
|
||||
ap.add_argument("--ma60-prev", type=float, default=None)
|
||||
ap.add_argument("--out", default=str(DEFAULT_OUT))
|
||||
args = ap.parse_args()
|
||||
|
||||
result = {
|
||||
"formula_id": "GOLDEN_CROSS_SIGNAL_V1",
|
||||
"golden_cross_today": golden_cross_today(args.ma20, args.ma20_prev, args.ma60, args.ma60_prev),
|
||||
"hard_constraint": "auxiliary_signal_only_not_buy_trigger",
|
||||
}
|
||||
out = Path(args.out)
|
||||
out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,39 @@
|
||||
#!/usr/bin/env python3
|
||||
"""STRONG_CLOSE_SIGNAL_V1 — spec/formulas/domains/entry.yaml.
|
||||
|
||||
governance/todo/technical_signals_p4_adoption_plan.yaml P4-2.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_OUT = ROOT / "Temp" / "strong_close_signal_v1.json"
|
||||
|
||||
|
||||
def evaluate_strong_close(close_price: float, high_price: float, low_price: float) -> dict:
|
||||
if high_price == low_price:
|
||||
return {"close_position_pct": None, "strong_close": None}
|
||||
close_position_pct = (close_price - low_price) / (high_price - low_price) * 100
|
||||
return {"close_position_pct": close_position_pct, "strong_close": close_position_pct >= 80}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--close", type=float, required=True)
|
||||
ap.add_argument("--high", type=float, required=True)
|
||||
ap.add_argument("--low", type=float, required=True)
|
||||
ap.add_argument("--out", default=str(DEFAULT_OUT))
|
||||
args = ap.parse_args()
|
||||
|
||||
result = {"formula_id": "STRONG_CLOSE_SIGNAL_V1", **evaluate_strong_close(args.close, args.high, args.low)}
|
||||
out = Path(args.out)
|
||||
out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,41 @@
|
||||
#!/usr/bin/env python3
|
||||
"""TREND_FILTER_GATE_V1 — spec/formulas/domains/entry.yaml.
|
||||
|
||||
governance/todo/technical_signals_p4_adoption_plan.yaml P4-7.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_OUT = ROOT / "Temp" / "trend_filter_gate_v1.json"
|
||||
|
||||
|
||||
def evaluate(close_price: float | None, ma120: float | None, ma120_prev: float | None) -> bool | None:
|
||||
if close_price is None or ma120 is None or ma120_prev is None:
|
||||
return None
|
||||
return close_price > ma120 and ma120 > ma120_prev
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--close", type=float, default=None)
|
||||
ap.add_argument("--ma120", type=float, default=None)
|
||||
ap.add_argument("--ma120-prev", type=float, default=None)
|
||||
ap.add_argument("--out", default=str(DEFAULT_OUT))
|
||||
args = ap.parse_args()
|
||||
|
||||
result = {
|
||||
"formula_id": "TREND_FILTER_GATE_V1",
|
||||
"trend_filter_pass": evaluate(args.close, args.ma120, args.ma120_prev),
|
||||
}
|
||||
out = Path(args.out)
|
||||
out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,53 @@
|
||||
#!/usr/bin/env python3
|
||||
"""VOLATILITY_EXPANSION_BREAKOUT_V1 — spec/formulas/domains/entry.yaml.
|
||||
|
||||
This signal does NOT authorize a buy by itself -- callers must separately confirm
|
||||
BREAKOUT_QUALITY_GATE_V2 != BLOCKED_LATE_CHASE. governance/todo/technical_signals_p4_adoption_plan.yaml P4-3.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_OUT = ROOT / "Temp" / "volatility_expansion_breakout_v1.json"
|
||||
|
||||
SQUEEZE_PERCENTILE_THRESHOLD = 20.0
|
||||
EXPANSION_RET_THRESHOLD_PCT = 3.0
|
||||
|
||||
|
||||
def squeeze_detected(bb_width_20d_percentile: float | None) -> bool | None:
|
||||
if bb_width_20d_percentile is None:
|
||||
return None
|
||||
return bb_width_20d_percentile <= SQUEEZE_PERCENTILE_THRESHOLD
|
||||
|
||||
|
||||
def evaluate(squeeze_detected_previous_day: bool | None, ret_1d: float | None) -> bool | None:
|
||||
if squeeze_detected_previous_day is None or ret_1d is None:
|
||||
return None
|
||||
return squeeze_detected_previous_day and ret_1d >= EXPANSION_RET_THRESHOLD_PCT
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--bb-width-20d-percentile-prev", type=float, default=None)
|
||||
ap.add_argument("--ret-1d", type=float, default=None)
|
||||
ap.add_argument("--out", default=str(DEFAULT_OUT))
|
||||
args = ap.parse_args()
|
||||
|
||||
prev_squeeze = squeeze_detected(args.bb_width_20d_percentile_prev)
|
||||
result = {
|
||||
"formula_id": "VOLATILITY_EXPANSION_BREAKOUT_V1",
|
||||
"squeeze_detected_previous_day": prev_squeeze,
|
||||
"volatility_expansion_breakout": evaluate(prev_squeeze, args.ret_1d),
|
||||
"hard_constraint": "requires_BREAKOUT_QUALITY_GATE_V2_pass_separately",
|
||||
}
|
||||
out = Path(args.out)
|
||||
out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user