yuxuanhui
57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
"""Application use case for one-stock historical Zhixing B1 evaluation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from datetime import date
|
|
|
|
from ..domain.models import SelectionEvaluation, StockHistory
|
|
from ..domain.ports import MarketDataReader, MarketDataReaderError
|
|
from ..domain.zhixing_b1 import ZhixingB1Strategy
|
|
|
|
|
|
class EvaluateZhixingB1:
|
|
"""Read one history, evaluate the pure strategy, and map read failures."""
|
|
|
|
def __init__(
|
|
self,
|
|
reader: MarketDataReader,
|
|
strategy: ZhixingB1Strategy | None = None,
|
|
) -> None:
|
|
"""Inject the market-data port and optionally a strategy instance."""
|
|
|
|
self.reader = reader
|
|
self.strategy = strategy or ZhixingB1Strategy()
|
|
|
|
def execute(self, ts_code: str, target_trade_date: date) -> SelectionEvaluation:
|
|
"""Evaluate ``ts_code`` on the exact requested trading date."""
|
|
|
|
try:
|
|
history = self.reader.load_history(ts_code, target_trade_date)
|
|
except MarketDataReaderError as exc:
|
|
return SelectionEvaluation(
|
|
ts_code=ts_code,
|
|
target_trade_date=target_trade_date,
|
|
status="data_error",
|
|
reason=str(exc),
|
|
)
|
|
return self.strategy.evaluate(history, target_trade_date)
|
|
|
|
def execute_history(
|
|
self,
|
|
history: StockHistory,
|
|
target_trade_date: date,
|
|
) -> SelectionEvaluation:
|
|
"""Evaluate an already loaded history for deterministic unit tests."""
|
|
|
|
return self.strategy.evaluate(history, target_trade_date)
|
|
|
|
def execute_histories(
|
|
self,
|
|
histories: Sequence[StockHistory],
|
|
target_trade_date: date,
|
|
) -> tuple[SelectionEvaluation, ...]:
|
|
"""Evaluate loaded histories without issuing one read per stock."""
|
|
|
|
return tuple(self.execute_history(history, target_trade_date) for history in histories)
|