387 lines
15 KiB
Python
387 lines
15 KiB
Python
from app.core.factor_roles import RISK, TRIGGER, factor_role, factor_roles_for_codes
|
||
from app.core.signal_direction import sanitize_factor_breakdown_for_side, sanitize_signals_for_side
|
||
from app.core.strategy_contract import StrategySignal, default_main_composite_signal
|
||
from app.core.strategy_registry import (
|
||
BREAKDOWN_RETEST_SHORT_1H_STRATEGY,
|
||
LONG_MOMENTUM_BREAKOUT_STRATEGY,
|
||
LONG_SECOND_WAVE_PULLBACK_STRATEGY,
|
||
MAIN_COMPOSITE_STRATEGY,
|
||
SHORT_BREAKDOWN_RETEST_STRATEGY,
|
||
is_strategy_allowed_for_side,
|
||
registered_strategy_codes,
|
||
strategy_direction,
|
||
strategy_label,
|
||
)
|
||
from app.db.recommendation_commands import create_recommendation
|
||
from app.db.paper_trading import _open_trade, _order_payload_from_rec
|
||
from app.db.strategy_signal_queries import insert_strategy_signal
|
||
from app.db.strategy_insights import evaluate_strategy_decision
|
||
from app.strategies.altcoin_breakout import (
|
||
build_long_momentum_breakout_signal,
|
||
build_long_second_wave_pullback_signal,
|
||
build_compression_breakout_4h_signal,
|
||
build_intraday_momentum_15m_signal,
|
||
build_volume_ignition_1h_signal,
|
||
)
|
||
from app.strategies.box_retest_4h import build_box_retest_1h_signal
|
||
from app.strategies.short_breakdown import build_breakdown_retest_short_1h_signal, detect_breakdown_retest_short_1h
|
||
|
||
|
||
def test_factor_roles_never_promote_unknown_to_trigger():
|
||
assert factor_role("box_breakout_pullback_4h") == TRIGGER
|
||
assert factor_role("box_breakout_pullback_1h") == TRIGGER
|
||
assert factor_role("false_breakout") == RISK
|
||
assert factor_role("new_unknown_factor") == "unknown"
|
||
assert factor_roles_for_codes(["box_breakout_pullback_4h", "new_unknown_factor"]) == {
|
||
"box_breakout_pullback_4h": "trigger",
|
||
"new_unknown_factor": "unknown",
|
||
}
|
||
|
||
|
||
def test_active_strategy_pool_only_contains_short_term_three_strategies():
|
||
signal = default_main_composite_signal(
|
||
symbol="AAA/USDT",
|
||
score=70,
|
||
signal_codes=["vp_fly_1h_current"],
|
||
entry_plan={"entry_action": "观察"},
|
||
).to_json_dict()
|
||
|
||
assert registered_strategy_codes() == [
|
||
LONG_MOMENTUM_BREAKOUT_STRATEGY,
|
||
LONG_SECOND_WAVE_PULLBACK_STRATEGY,
|
||
SHORT_BREAKDOWN_RETEST_STRATEGY,
|
||
]
|
||
assert signal["strategy_code"] == LONG_MOMENTUM_BREAKOUT_STRATEGY
|
||
assert signal["strategy_name"] == "多头动量启动"
|
||
assert signal["factor_roles"]["vp_fly_1h_current"] == "trigger"
|
||
assert strategy_label(LONG_MOMENTUM_BREAKOUT_STRATEGY) == "多头动量启动"
|
||
assert strategy_label(LONG_SECOND_WAVE_PULLBACK_STRATEGY) == "多头二波回踩"
|
||
assert strategy_label(SHORT_BREAKDOWN_RETEST_STRATEGY) == "空头破位反抽"
|
||
assert strategy_direction(LONG_MOMENTUM_BREAKOUT_STRATEGY) == "long"
|
||
assert strategy_direction(BREAKDOWN_RETEST_SHORT_1H_STRATEGY) == "short"
|
||
assert is_strategy_allowed_for_side(MAIN_COMPOSITE_STRATEGY, "short") is False
|
||
assert is_strategy_allowed_for_side(LONG_MOMENTUM_BREAKOUT_STRATEGY, "short") is False
|
||
assert is_strategy_allowed_for_side(BREAKDOWN_RETEST_SHORT_1H_STRATEGY, "short") is True
|
||
|
||
|
||
def test_long_momentum_breakout_strategy_builds_independent_signal():
|
||
signal = build_long_momentum_breakout_signal(
|
||
symbol="VOL/USDT",
|
||
result={
|
||
"score": 8,
|
||
"signals": ["1H量价齐飞 · 连续放量", "15min即刻入场"],
|
||
"trigger_context": {"current_triggers": ["15m突破"], "trigger_status": "current"},
|
||
"entry_plan": {"entry_action": "可即刻买入"},
|
||
},
|
||
entry_plan={"entry_action": "可即刻买入", "entry_price": 1.0},
|
||
)
|
||
payload = signal.to_json_dict()
|
||
|
||
assert payload["strategy_code"] == LONG_MOMENTUM_BREAKOUT_STRATEGY
|
||
assert payload["status"] == "candidate"
|
||
assert payload["trigger"]["factor_code"] == "momentum_breakout_15m_1h"
|
||
assert payload["factor_roles"]["momentum_breakout_15m_1h"] == "trigger"
|
||
|
||
|
||
def test_legacy_strategy_builders_no_longer_emit_signals():
|
||
assert build_volume_ignition_1h_signal(
|
||
symbol="OLD/USDT",
|
||
result={"score": 9, "signals": ["1H量价齐飞", "15min即刻入场"], "trigger_context": {"current_triggers": ["15m"]}},
|
||
entry_plan={"entry_action": "可即刻买入"},
|
||
) is None
|
||
assert build_compression_breakout_4h_signal(
|
||
symbol="OLD/USDT",
|
||
result={"score": 9, "signals": ["4H静K压缩，突破箱体上沿"]},
|
||
entry_plan={"entry_action": "等回踩"},
|
||
) is None
|
||
assert build_intraday_momentum_15m_signal(
|
||
symbol="OLD/USDT",
|
||
result={"score": 9, "signals": ["15min强突破"], "trigger_context": {"current_triggers": ["15m"]}},
|
||
entry_plan={"entry_action": "可即刻买入"},
|
||
) is None
|
||
|
||
|
||
def test_long_strategy_cannot_emit_short_signal():
|
||
import pytest
|
||
|
||
with pytest.raises(ValueError):
|
||
StrategySignal(
|
||
strategy_code=LONG_MOMENTUM_BREAKOUT_STRATEGY,
|
||
symbol="BAD/USDT",
|
||
direction="short",
|
||
factor_roles={"vp_fly_1h_current": "trigger"},
|
||
)
|
||
|
||
|
||
def test_short_direction_filters_bullish_supporting_evidence():
|
||
signals = [
|
||
"大户偏多(73%)",
|
||
"BTC回调中独立走强",
|
||
"板块联动: Layer1 龙头TON/USDT",
|
||
"1H箱体突破回踩",
|
||
"1H破位反抽做空",
|
||
"破位质量高",
|
||
]
|
||
clean, removed = sanitize_signals_for_side(signals, "short")
|
||
|
||
assert "1H破位反抽做空" in clean
|
||
assert "破位质量高" in clean
|
||
assert all("大户偏多" not in item for item in clean)
|
||
assert all("BTC回调中独立走强" not in item for item in clean)
|
||
assert any("板块联动" in item for item in removed)
|
||
|
||
|
||
def test_short_factor_breakdown_removes_long_only_positive_factors():
|
||
summary = {
|
||
"items": [
|
||
{"factor_code": "top_trader_long", "factor_group": "positioning", "score_delta": 1},
|
||
{"factor_code": "sector_rotation", "factor_group": "narrative", "score_delta": 2},
|
||
{"factor_code": "box_breakout_pullback_1h", "factor_group": "structure", "score_delta": 6},
|
||
{"factor_code": "breakdown_retest_1h_short", "factor_group": "structure", "score_delta": 7},
|
||
{"factor_code": "funding_positive_risk", "factor_group": "risk", "score_delta": -3},
|
||
]
|
||
}
|
||
|
||
clean, removed = sanitize_factor_breakdown_for_side(summary, "short")
|
||
codes = [item["factor_code"] for item in clean["items"]]
|
||
|
||
assert "breakdown_retest_1h_short" in codes
|
||
assert "funding_positive_risk" in codes
|
||
assert "top_trader_long" not in codes
|
||
assert "sector_rotation" not in codes
|
||
assert "box_breakout_pullback_1h" not in codes
|
||
assert clean["total_delta"] == 4
|
||
assert {item["factor_code"] for item in removed} == {
|
||
"top_trader_long",
|
||
"sector_rotation",
|
||
"box_breakout_pullback_1h",
|
||
}
|
||
|
||
|
||
def test_second_wave_strategy_requires_first_wave_and_pullback_context():
|
||
signal = build_long_second_wave_pullback_signal(
|
||
symbol="QUIET/USDT",
|
||
result={"score": 6, "signals": ["24h强势榜异动", "1H箱体突破回踩"], "entry_plan": {"entry_action": "等回踩"}},
|
||
entry_plan={"entry_action": "等回踩", "entry_price": 1.0},
|
||
)
|
||
payload = signal.to_json_dict()
|
||
|
||
assert payload["strategy_code"] == LONG_SECOND_WAVE_PULLBACK_STRATEGY
|
||
assert payload["status"] == "candidate"
|
||
assert payload["factor_roles"]["second_wave_pullback_1h"] == "trigger"
|
||
assert build_long_second_wave_pullback_signal(
|
||
symbol="NOBOX/USDT",
|
||
result={"score": 6, "signals": ["1H量价齐飞"], "entry_plan": {"entry_action": "可即刻买入"}},
|
||
entry_plan={"entry_action": "可即刻买入"},
|
||
) is None
|
||
|
||
|
||
def test_long_momentum_strategy_requires_current_trigger():
|
||
stale = build_long_momentum_breakout_signal(
|
||
symbol="FAST/USDT",
|
||
result={"score": 7, "signals": ["15m短周期启动", "1H量价齐飞"], "entry_plan": {"entry_action": "可即刻买入"}},
|
||
entry_plan={"entry_action": "可即刻买入"},
|
||
).to_json_dict()
|
||
fresh = build_long_momentum_breakout_signal(
|
||
symbol="FAST/USDT",
|
||
result={
|
||
"score": 7,
|
||
"signals": ["15min强突破", "1H量价齐飞"],
|
||
"trigger_context": {"current_triggers": ["15m突破"], "trigger_status": "current"},
|
||
"entry_plan": {"entry_action": "可即刻买入"},
|
||
},
|
||
entry_plan={"entry_action": "可即刻买入"},
|
||
).to_json_dict()
|
||
|
||
assert stale["status"] == "observe"
|
||
assert "缺少当前低周期触发" in stale["risk_plan"]["risk_reasons"]
|
||
assert fresh["status"] == "candidate"
|
||
|
||
|
||
def test_breakdown_retest_short_strategy_is_independent_short_signal():
|
||
signal = build_breakdown_retest_short_1h_signal(
|
||
symbol="WEAK/USDT",
|
||
current_price=0.98,
|
||
detection={
|
||
"detected": True,
|
||
"score": 8,
|
||
"breakdown_level": 1.0,
|
||
"retest_zone": 1.0,
|
||
"stop_level": 1.04,
|
||
"target_1": 0.9,
|
||
"quality": "优质",
|
||
"retest_rejected": True,
|
||
"relative_weakness": True,
|
||
},
|
||
market_regime={"risk_level": "high", "regime": "risk_off"},
|
||
).to_json_dict()
|
||
|
||
assert signal["strategy_code"] == BREAKDOWN_RETEST_SHORT_1H_STRATEGY
|
||
assert signal["direction"] == "short"
|
||
assert signal["entry_plan"]["side"] == "short"
|
||
assert signal["status"] == "candidate"
|
||
assert signal["factor_roles"]["breakdown_retest_1h_short"] == "trigger"
|
||
|
||
|
||
def test_short_breakdown_detector_exposes_numeric_quality_score():
|
||
import pandas as pd
|
||
|
||
rows = []
|
||
price = 1.0
|
||
for i in range(64):
|
||
rows.append({"timestamp": i, "open": price, "high": 1.02, "low": 0.98, "close": price, "volume": 100})
|
||
rows.extend([
|
||
{"timestamp": 64, "open": 1.0, "high": 1.01, "low": 0.95, "close": 0.97, "volume": 180},
|
||
{"timestamp": 65, "open": 0.97, "high": 1.01, "low": 0.96, "close": 0.99, "volume": 140},
|
||
{"timestamp": 66, "open": 0.99, "high": 1.015, "low": 0.955, "close": 0.965, "volume": 180},
|
||
{"timestamp": 67, "open": 0.965, "high": 0.98, "low": 0.94, "close": 0.955, "volume": 200},
|
||
])
|
||
result = detect_breakdown_retest_short_1h(pd.DataFrame(rows), change_24h=-4)
|
||
|
||
assert result["detected"] is True
|
||
assert result["quality"] in {"良好", "优质"}
|
||
assert isinstance(result["quality_score"], int)
|
||
assert result["quality_score"] >= 5
|
||
|
||
|
||
def test_strategy_evaluation_recommends_promote_or_pause():
|
||
strong = evaluate_strategy_decision({
|
||
"signal_count": 24,
|
||
"opportunity_count": 16,
|
||
"trade_count": 8,
|
||
"closed_trade_count": 8,
|
||
"win_rate_pct": 62.5,
|
||
"avg_realized_pnl_pct": 3.2,
|
||
"realized_pnl_usdt": 180,
|
||
"worst_pnl_pct": -3.5,
|
||
"order_fill_rate_pct": 45,
|
||
"trade_conversion_pct": 50,
|
||
})
|
||
weak = evaluate_strategy_decision({
|
||
"signal_count": 24,
|
||
"opportunity_count": 16,
|
||
"trade_count": 8,
|
||
"closed_trade_count": 8,
|
||
"win_rate_pct": 25,
|
||
"avg_realized_pnl_pct": -2.5,
|
||
"realized_pnl_usdt": -120,
|
||
"worst_pnl_pct": -9,
|
||
"order_fill_rate_pct": 20,
|
||
"trade_conversion_pct": 50,
|
||
})
|
||
unfilled = evaluate_strategy_decision({
|
||
"signal_count": 18,
|
||
"opportunity_count": 12,
|
||
"trade_count": 0,
|
||
"closed_trade_count": 0,
|
||
})
|
||
|
||
assert strong["decision"] == "promote"
|
||
assert strong["evaluation_score"] > weak["evaluation_score"]
|
||
assert weak["decision"] == "pause"
|
||
assert unfilled["decision"] == "review_entry_gate"
|
||
|
||
|
||
def test_strategy_signal_insert_and_recommendation_lineage(pg_conn):
|
||
signal = insert_strategy_signal(
|
||
StrategySignal(
|
||
strategy_code=LONG_SECOND_WAVE_PULLBACK_STRATEGY,
|
||
symbol="BOX/USDT",
|
||
score=10,
|
||
confidence=80,
|
||
trigger={"factor_code": "second_wave_pullback_1h"},
|
||
factor_roles={"second_wave_pullback_1h": "trigger"},
|
||
entry_plan={"entry_action": "等回踩", "entry_price": 1.0},
|
||
)
|
||
)
|
||
|
||
rec_id = create_recommendation(
|
||
symbol="BOX/USDT",
|
||
rec_state="爆发",
|
||
rec_score=30,
|
||
entry_price=1.0,
|
||
stop_loss=0.94,
|
||
tp1=1.12,
|
||
signals=["4H箱体突破回踩(箱体上沿 $1, 量2x)"],
|
||
entry_plan={"entry_action": "等回踩", "entry_price": 1.0, "stop_loss": 0.94, "tp1": 1.12},
|
||
strategy_code=signal["strategy_code"],
|
||
strategy_signal_id=signal["strategy_signal_id"],
|
||
strategy_snapshot=signal,
|
||
factor_roles=signal["factor_roles"],
|
||
)
|
||
row = pg_conn.execute("SELECT * FROM recommendation WHERE id=%s", (rec_id,)).fetchone()
|
||
|
||
assert row["strategy_code"] == LONG_SECOND_WAVE_PULLBACK_STRATEGY
|
||
assert row["strategy_signal_id"] == signal["strategy_signal_id"]
|
||
assert "second_wave_pullback_1h" in row["factor_roles_json"]
|
||
|
||
|
||
def test_box_retest_strategy_preserves_zero_age_as_fresh():
|
||
signal = build_box_retest_1h_signal(
|
||
symbol="FRESH/USDT",
|
||
current_price=1.01,
|
||
detection={
|
||
"detected": True,
|
||
"score": 10,
|
||
"entry_zone": 1.0,
|
||
"stop_level": 0.94,
|
||
"quality": "优质",
|
||
"pullback_age_bars": 0,
|
||
},
|
||
market_regime={"regime": "altcoin_rotation", "risk_level": "medium"},
|
||
)
|
||
assert signal is None
|
||
|
||
|
||
def test_paper_order_and_trade_inherit_strategy_lineage(pg_conn):
|
||
rec = {
|
||
"id": 1,
|
||
"symbol": "BOX/USDT",
|
||
"rec_score": 100,
|
||
"entry_price": 1.0,
|
||
"stop_loss": 0.94,
|
||
"tp1": 1.12,
|
||
"tp2": 1.2,
|
||
"execution_status": "buy_now",
|
||
"action_status": "可即刻买入",
|
||
"strategy_version": "v-test",
|
||
"strategy_code": LONG_SECOND_WAVE_PULLBACK_STRATEGY,
|
||
"strategy_signal_id": 42,
|
||
"strategy_snapshot_json": '{"strategy_code":"long_second_wave_pullback_1h_v1"}',
|
||
"factor_roles_json": '{"second_wave_pullback_1h":"trigger"}',
|
||
"entry_plan": {"entry_action": "可即刻买入", "entry_price": 1.0, "stop_loss": 0.94, "tp1": 1.12},
|
||
}
|
||
|
||
payload = _order_payload_from_rec(rec, 1.01, "2026-05-27T00:00:00", {"trade_notional_usdt": 100})
|
||
assert payload["strategy_code"] == LONG_SECOND_WAVE_PULLBACK_STRATEGY
|
||
assert payload["strategy_signal_id"] == 42
|
||
|
||
result = _open_trade(
|
||
pg_conn,
|
||
rec,
|
||
1.0,
|
||
"2026-05-27T00:00:00",
|
||
config={
|
||
"enabled": True,
|
||
"trade_notional_usdt": 100,
|
||
"trade_leverage": 1,
|
||
"account_equity_usdt": 20000,
|
||
"fee_rate": 0,
|
||
"min_rec_score": 0,
|
||
"min_rr": 0,
|
||
"max_stop_loss_leverage_risk_pct": 999,
|
||
"max_cumulative_leverage": 999,
|
||
"max_account_drawdown_pause_pct": 0,
|
||
"weak_entry_pause": {"enabled": False},
|
||
},
|
||
push_open_card=False,
|
||
)
|
||
assert result["opened"] is True
|
||
row = pg_conn.execute("SELECT * FROM paper_trades WHERE id=%s", (result["trade_id"],)).fetchone()
|
||
event = pg_conn.execute("SELECT * FROM paper_trade_events WHERE trade_id=%s", (result["trade_id"],)).fetchone()
|
||
|
||
assert row["strategy_code"] == LONG_SECOND_WAVE_PULLBACK_STRATEGY
|
||
assert row["strategy_signal_id"] == 42
|
||
assert event["strategy_code"] == LONG_SECOND_WAVE_PULLBACK_STRATEGY
|
||
assert strategy_label(row["strategy_code"]) == "多头二波回踩"
|