jackyu66gitandCursor
03851d2247
新增规则驱动的月/周/日结构识别、决策融合与交易计划，提供扫描 API、本地 K 线（成交量/MACD/吸筹区间标注）及回填调度；K 线无起始日时默认取最近 N 根。 Co-authored-by: Cursor <cursoragent@cursor.com>
95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
"""Plan gate + insufficient TF fallback tests."""
|
|
|
|
from datetime import date, timedelta
|
|
|
|
from ashare_dp.domain.wyckoff import DecisionSignal, EngineResult, OHLCVFrame, WyckoffCycle
|
|
from ashare_dp.wyckoff.cycle import CycleEngine
|
|
from ashare_dp.wyckoff.decision import DecisionEngine
|
|
from ashare_dp.wyckoff.features import FeatureEngine
|
|
from ashare_dp.wyckoff.plan import PlanEngine
|
|
from ashare_dp.wyckoff.pipeline import analyze_symbol
|
|
from ashare_dp.wyckoff.phase import PhaseEngine
|
|
from ashare_dp.wyckoff.event import EventEngine
|
|
from ashare_dp.wyckoff.signal import SignalEngine
|
|
|
|
|
|
def _er(name, payload, confidence=80.0, score=80.0):
|
|
return EngineResult(name=name, confidence=confidence, score=score, payload=payload)
|
|
|
|
|
|
def test_plan_no_entry_on_watch_even_if_spring_event():
|
|
plan = PlanEngine()
|
|
feat = _er("Feature", {"close": 10.0, "atr": 0.3, "swing_low": 9.0, "swing_high": 11.0, "range_high": 11.0})
|
|
decision = _er(
|
|
"Decision",
|
|
{
|
|
"decision_signal": DecisionSignal.WATCH.value,
|
|
"d_event": "Spring",
|
|
},
|
|
confidence=90,
|
|
score=50,
|
|
)
|
|
out = plan.run(feat, decision)
|
|
assert out.payload["entry"] is None
|
|
assert out.payload["stop"] is None
|
|
|
|
|
|
def test_plan_entry_on_buy():
|
|
plan = PlanEngine()
|
|
feat = _er("Feature", {"close": 10.0, "atr": 0.3, "swing_low": 9.0, "swing_high": 11.0, "range_high": 11.0})
|
|
decision = _er("Decision", {"decision_signal": DecisionSignal.BUY.value, "d_event": "Spring"})
|
|
out = plan.run(feat, decision)
|
|
assert out.payload["entry"] == 10.0
|
|
assert out.payload["stop"] is not None
|
|
|
|
|
|
def test_feature_insufficient_for_short_monthly():
|
|
fe = FeatureEngine()
|
|
base = date(2024, 1, 1)
|
|
n = 10
|
|
frame = OHLCVFrame(
|
|
ts_code="000001.SZ",
|
|
timeframe="1M",
|
|
trade_dates=[base + timedelta(days=30 * i) for i in range(n)],
|
|
open=[10.0] * n,
|
|
high=[11.0] * n,
|
|
low=[9.0] * n,
|
|
close=[10.0] * n,
|
|
volume=[1e6] * n,
|
|
)
|
|
out = fe.run(frame, "1M")
|
|
assert out.payload["insufficient"] is True
|
|
cyc = CycleEngine().run(out, "1M")
|
|
assert cyc.payload["cycle"] == WyckoffCycle.UNKNOWN.value
|
|
|
|
|
|
def test_pipeline_does_not_borrow_daily_as_monthly():
|
|
"""Daily-only data → monthly cycle Unknown, not inferred from daily."""
|
|
base = date(2024, 1, 1)
|
|
n = 120
|
|
closes = [100 + i * 0.4 for i in range(n)]
|
|
daily = OHLCVFrame(
|
|
ts_code="000001.SZ",
|
|
timeframe="1d",
|
|
trade_dates=[base + timedelta(days=i) for i in range(n)],
|
|
open=closes,
|
|
high=[c * 1.01 for c in closes],
|
|
low=[c * 0.99 for c in closes],
|
|
close=closes,
|
|
volume=[1e6] * n,
|
|
)
|
|
result = analyze_symbol(
|
|
daily,
|
|
None,
|
|
None,
|
|
feature_eng=FeatureEngine(),
|
|
cycle_eng=CycleEngine(),
|
|
phase_eng=PhaseEngine(),
|
|
event_eng=EventEngine(),
|
|
signal_eng=SignalEngine(),
|
|
decision_eng=DecisionEngine(),
|
|
plan_eng=PlanEngine(),
|
|
)
|
|
assert result["f_m"].payload.get("insufficient") is True
|
|
assert result["c_m"].payload["cycle"] == WyckoffCycle.UNKNOWN.value
|