mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 19:04:15 +08:00
intfoo
3e6f9bcfb7
- kline.sync_minute_single 对指数 symbol 显式 400 (防污染 kline_minute) - kline.sync_minute 全市场 universe 剔除指数 symbol - kline.get_minute 响应新增 asset_type 字段 (3 处 return) - monitor_rules._reconcile_index_asset_type 纠正误存为 stock 的指数规则 (应用于 save_rule/list_rules/_sync_engine, 混合池不动) - 文案资产中立化: "个股信号→信号" "指定股票→指定标的" (options label + 校验报错) - 测试: reconcile 5 断言 + sync_minute_single 拒指数 400 + 分时报错断言同步
109 lines
3.9 KiB
Python
109 lines
3.9 KiB
Python
"""指数监控规则校验测试。"""
|
|
import pytest
|
|
|
|
from app.strategy import monitor_rules
|
|
|
|
|
|
def _index_rule(rid="r_idx", **over):
|
|
rule = {
|
|
"id": rid, "name": rid, "type": "signal", "asset_type": "index",
|
|
"scope": "symbols", "symbols": ["000001.SH"], "logic": "and",
|
|
"conditions": [{"field": "rsi_14", "op": "<", "value": 30}],
|
|
"cooldown_seconds": 0, "enabled": True,
|
|
}
|
|
rule.update(over)
|
|
return rule
|
|
|
|
|
|
def test_index_signal_price_allowed():
|
|
monitor_rules.validate(_index_rule())
|
|
monitor_rules.validate(_index_rule(type="price"))
|
|
|
|
|
|
def test_index_strategy_rejected():
|
|
with pytest.raises(ValueError, match="指数"):
|
|
monitor_rules.validate(_index_rule(type="strategy", strategy_id="s1"))
|
|
|
|
|
|
def test_index_market_rejected():
|
|
with pytest.raises(ValueError, match="指数"):
|
|
monitor_rules.validate(_index_rule(type="market"))
|
|
|
|
|
|
def test_index_scope_all_rejected():
|
|
with pytest.raises(ValueError, match="指数"):
|
|
monitor_rules.validate(_index_rule(scope="all", symbols=[]))
|
|
|
|
|
|
def test_index_intraday_signal_rejected():
|
|
with pytest.raises(ValueError, match="分时"):
|
|
monitor_rules.validate(_index_rule(
|
|
conditions=[{"field": "signal_intraday_avg_cross_up", "op": "truth"}],
|
|
))
|
|
|
|
|
|
# ---- Task 7: B5 监控指数评估轮 ----
|
|
|
|
def _signal_rule(rid, asset_type, sym):
|
|
return {
|
|
"id": rid, "name": rid, "type": "signal", "asset_type": asset_type,
|
|
"scope": "symbols", "symbols": [sym], "logic": "and",
|
|
"conditions": [{"field": "rsi_14", "op": "<", "value": 100}],
|
|
"cooldown_seconds": 0, "enabled": True,
|
|
}
|
|
|
|
|
|
def test_evaluate_index_round_triggers_and_isolates():
|
|
"""指数轮只评估指数规则, 且不触碰策略结果缓存。"""
|
|
import polars as pl
|
|
from app.strategy.monitor import MonitorRuleEngine
|
|
|
|
eng = MonitorRuleEngine()
|
|
eng.set_rules([_signal_rule("r_idx", "index", "000001.SH"),
|
|
_signal_rule("r_stock", "stock", "000001.SH")])
|
|
eng.set_name_map({"000001.SH": "上证指数"})
|
|
df = pl.DataFrame({"symbol": ["000001.SH"], "close": [3000.0],
|
|
"change_pct": [0.01], "rsi_14": [40.0]})
|
|
|
|
events = eng.evaluate(df, asset_type="index", reset_strategy_results=False)
|
|
assert any(e["rule_id"] == "r_idx" for e in events)
|
|
assert all(e["rule_id"] != "r_stock" for e in events)
|
|
assert events[0]["name"] == "上证指数"
|
|
assert eng.latest_strategy_results() == {} # 策略结果缓存未被触碰
|
|
|
|
|
|
# ---- 资产类型纠正: 误存为 stock 的指数规则 ----
|
|
|
|
class _FakeRepo:
|
|
def resolve_asset_type(self, symbol):
|
|
return {"000001.SH": "index"}.get(symbol, "stock")
|
|
|
|
|
|
def test_reconcile_index_asset_type_corrects_index_only_rule():
|
|
from app.api.monitor_rules import _reconcile_index_asset_type
|
|
|
|
rule = {"asset_type": "stock", "scope": "symbols", "symbols": ["000001.SH"]}
|
|
assert _reconcile_index_asset_type(rule, _FakeRepo())["asset_type"] == "index"
|
|
|
|
|
|
def test_reconcile_index_asset_type_keeps_stock_and_mixed():
|
|
from app.api.monitor_rules import _reconcile_index_asset_type
|
|
|
|
repo = _FakeRepo()
|
|
# 纯股票 → 不动
|
|
assert _reconcile_index_asset_type(
|
|
{"asset_type": "stock", "scope": "symbols", "symbols": ["600000.SH"]}, repo,
|
|
)["asset_type"] == "stock"
|
|
# 股票+指数混合 → 不动 (asset_type 语义覆盖整条规则)
|
|
assert _reconcile_index_asset_type(
|
|
{"asset_type": "stock", "scope": "symbols", "symbols": ["000001.SH", "600000.SH"]}, repo,
|
|
)["asset_type"] == "stock"
|
|
# 已是 index → 不动
|
|
assert _reconcile_index_asset_type(
|
|
{"asset_type": "index", "scope": "symbols", "symbols": ["000001.SH"]}, repo,
|
|
)["asset_type"] == "index"
|
|
# 非 symbols 范围 → 不动
|
|
assert _reconcile_index_asset_type(
|
|
{"asset_type": "stock", "scope": "all", "symbols": []}, repo,
|
|
)["asset_type"] == "stock"
|