95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
|
"""Behavior tests for explicit-date Zhixing B1 evaluation."""
|
||
|
|
|
||
|
|
from datetime import date, timedelta
|
||
|
|
|
||
|
|
import pandas as pd
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from zhixing_server.modules.selection.domain import zhixing_b1
|
||
|
|
from zhixing_server.modules.selection.domain.models import SelectionBar, StockHistory
|
||
|
|
from zhixing_server.modules.selection.domain.zhixing_b1 import (
|
||
|
|
MINIMUM_HISTORY,
|
||
|
|
ZHIXING_B1_SIGNAL_ORDER,
|
||
|
|
ZhixingB1Strategy,
|
||
|
|
compute_signal_masks,
|
||
|
|
prepare_zhixing_b1_indicators,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def make_history(count: int = MINIMUM_HISTORY, code: str = "000001.SZ") -> StockHistory:
|
||
|
|
bars = tuple(
|
||
|
|
SelectionBar(
|
||
|
|
trade_date=date(2020, 1, 1) + timedelta(days=index),
|
||
|
|
open=10.0 + index * 0.02,
|
||
|
|
high=10.2 + index * 0.02,
|
||
|
|
low=9.9 + index * 0.02,
|
||
|
|
close=10.1 + index * 0.02,
|
||
|
|
volume=1000.0 + (index % 7) * 30,
|
||
|
|
)
|
||
|
|
for index in range(count)
|
||
|
|
)
|
||
|
|
return StockHistory(ts_code=code, name="测试股票", bars=bars)
|
||
|
|
|
||
|
|
|
||
|
|
def test_strategy_has_seven_stable_categories_and_prepared_masks() -> None:
|
||
|
|
history = make_history()
|
||
|
|
frame = pd.DataFrame(
|
||
|
|
{
|
||
|
|
"open": [bar.open for bar in history.bars],
|
||
|
|
"high": [bar.high for bar in history.bars],
|
||
|
|
"low": [bar.low for bar in history.bars],
|
||
|
|
"close": [bar.close for bar in history.bars],
|
||
|
|
"volume": [bar.volume for bar in history.bars],
|
||
|
|
}
|
||
|
|
)
|
||
|
|
prepared = prepare_zhixing_b1_indicators(frame, history.ts_code)
|
||
|
|
masks = compute_signal_masks(prepared)
|
||
|
|
|
||
|
|
assert tuple(masks) == ZHIXING_B1_SIGNAL_ORDER
|
||
|
|
assert all(mask.dtype == bool for mask in masks.values())
|
||
|
|
assert all(len(mask) == len(history.bars) for mask in masks.values())
|
||
|
|
|
||
|
|
|
||
|
|
def test_strategy_explicit_target_ignores_future_rows() -> None:
|
||
|
|
history = make_history()
|
||
|
|
target = history.bars[-1].trade_date
|
||
|
|
future = SelectionBar(
|
||
|
|
trade_date=target + timedelta(days=1),
|
||
|
|
open=1.0,
|
||
|
|
high=100.0,
|
||
|
|
low=0.5,
|
||
|
|
close=99.0,
|
||
|
|
volume=1_000_000.0,
|
||
|
|
)
|
||
|
|
with_future = StockHistory(history.ts_code, history.name, history.bars + (future,))
|
||
|
|
|
||
|
|
strategy = ZhixingB1Strategy()
|
||
|
|
assert strategy.evaluate(with_future, target) == strategy.evaluate(history, target)
|
||
|
|
|
||
|
|
|
||
|
|
def test_strategy_returns_missing_and_warmup_states() -> None:
|
||
|
|
strategy = ZhixingB1Strategy()
|
||
|
|
history = make_history(MINIMUM_HISTORY - 1)
|
||
|
|
target = history.bars[-1].trade_date
|
||
|
|
assert strategy.evaluate(history, target).status == "insufficient_history"
|
||
|
|
assert strategy.evaluate(history, target + timedelta(days=1)).status == "missing_target_bar"
|
||
|
|
|
||
|
|
|
||
|
|
def test_strategy_keeps_all_same_day_subsignals_in_priority_order(
|
||
|
|
monkeypatch: pytest.MonkeyPatch,
|
||
|
|
) -> None:
|
||
|
|
history = make_history()
|
||
|
|
target = history.bars[-1].trade_date
|
||
|
|
|
||
|
|
def all_masks(frame: pd.DataFrame) -> dict[zhixing_b1.ZhixingB1Category, pd.Series]:
|
||
|
|
return {
|
||
|
|
category: pd.Series(True, index=frame.index) for category in ZHIXING_B1_SIGNAL_ORDER
|
||
|
|
}
|
||
|
|
|
||
|
|
monkeypatch.setattr(zhixing_b1, "compute_signal_masks", all_masks)
|
||
|
|
result = ZhixingB1Strategy().evaluate(history, target)
|
||
|
|
|
||
|
|
assert result.status == "selected"
|
||
|
|
assert tuple(signal.category for signal in result.signals) == ZHIXING_B1_SIGNAL_ORDER
|
||
|
|
assert len({signal.identity for signal in result.signals}) == 7
|