yuxuanhui
68282f5d46
- pin the selection sector vocabulary to industry: the /sectors endpoint no longer accepts sector_type, and the port, adapter, and use case all resolve counts and member codes with sector_type="industry" - relabel the results filter to 细分行业 and drop concept-type plumbing from the frontend API, query keys, and types - show only 细分行业 in the signal detail panel; remove the concept-board chips and keep the snapshot-date tooltip on the industry line - update backend and frontend tests to the industry vocabulary and record the revised scope in the task PRD
775 lines
30 KiB
Python
775 lines
30 KiB
Python
"""Application orchestration for persisted whole-universe strategy runs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from collections import Counter
|
|
from collections.abc import Callable, Mapping, Sequence
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from dataclasses import dataclass, replace
|
|
from datetime import date
|
|
from typing import Protocol, cast
|
|
|
|
from ..domain.models import SelectionEvaluation, SelectionStrategyName, StockHistory
|
|
from ..domain.pattern_scoring import (
|
|
PatternCase,
|
|
PatternCaseLibraryLoader,
|
|
PatternScore,
|
|
PatternScorer,
|
|
)
|
|
from ..domain.runs import (
|
|
BatchSelectionRunStore,
|
|
BatchSelectionUniverseReader,
|
|
SelectionExecutionSource,
|
|
SelectionRerunRequired,
|
|
SelectionResultQuery,
|
|
SelectionRun,
|
|
SelectionRunInProgress,
|
|
SelectionRunItem,
|
|
SelectionRunStatus,
|
|
SelectionRunStore,
|
|
SelectionSectorCount,
|
|
SelectionSectorMembership,
|
|
SelectionSectorReader,
|
|
SelectionStock,
|
|
SelectionUniverseReader,
|
|
)
|
|
from .evaluate import EvaluateZhixingB1
|
|
|
|
# Selection runs are started by the ASGI service in production. A child of
|
|
# Uvicorn's configured logger keeps INFO diagnostics visible in container logs.
|
|
logger = logging.getLogger("uvicorn.error.zhixing.selection.run")
|
|
StrategyName = SelectionStrategyName
|
|
_FAILURE_STATUSES = {
|
|
"insufficient_history",
|
|
"missing_target_bar",
|
|
"missing_turnover_rate",
|
|
"data_error",
|
|
}
|
|
|
|
|
|
class SelectionEvaluator(Protocol):
|
|
"""Minimal single-stock evaluator required by the batch orchestrator."""
|
|
|
|
def execute(self, ts_code: str, target_trade_date: date) -> SelectionEvaluation: ...
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PreparedSelectionRun:
|
|
"""A claimed run and its immutable market-data source snapshot."""
|
|
|
|
run: SelectionRun
|
|
source: SelectionExecutionSource
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SelectionSectorAggregates:
|
|
"""A run summary plus its selected stocks' sector membership counts."""
|
|
|
|
run: SelectionRun | None
|
|
snapshot_trade_date: date | None
|
|
sector_type: str
|
|
sectors: tuple[SelectionSectorCount, ...]
|
|
|
|
|
|
class RunZhixingB1:
|
|
"""Prepare, execute, and query persisted selection strategy batches.
|
|
|
|
The historical class name remains as a compatibility seam for existing
|
|
composition and tests while strategy routing is now explicit.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
reader: SelectionUniverseReader,
|
|
store: SelectionRunStore,
|
|
evaluator: SelectionEvaluator | None = None,
|
|
evaluators: Mapping[StrategyName, SelectionEvaluator] | None = None,
|
|
pattern_case_loader: PatternCaseLibraryLoader | None = None,
|
|
pattern_scorer: PatternScorer | None = None,
|
|
sector_reader: SelectionSectorReader | None = None,
|
|
*,
|
|
pattern_scoring_enabled: bool = False,
|
|
max_workers: int = 4,
|
|
batch_size: int = 200,
|
|
) -> None:
|
|
"""Inject storage ports and configure bounded chunk execution."""
|
|
|
|
if max_workers < 1:
|
|
raise ValueError("max_workers must be at least 1")
|
|
if batch_size < 1:
|
|
raise ValueError("batch_size must be at least 1")
|
|
self.reader = reader
|
|
self.store = store
|
|
self.evaluator = evaluator or EvaluateZhixingB1(reader)
|
|
self.evaluators: dict[StrategyName, SelectionEvaluator] = {
|
|
"zhixing_b1": self.evaluator,
|
|
}
|
|
if evaluators is not None:
|
|
self.evaluators.update(evaluators)
|
|
self.pattern_case_loader = pattern_case_loader
|
|
self.pattern_scorer = pattern_scorer
|
|
self.sector_reader = sector_reader
|
|
self.pattern_scoring_enabled = pattern_scoring_enabled
|
|
self.max_workers = max_workers
|
|
self.batch_size = batch_size
|
|
|
|
def prepare(
|
|
self,
|
|
strategy: StrategyName,
|
|
target_trade_date: date,
|
|
*,
|
|
rerun: bool,
|
|
) -> PreparedSelectionRun:
|
|
"""Validate source eligibility before claiming the rerunnable key."""
|
|
|
|
logger.info(
|
|
"selection_run_prepare_started strategy=%s target_trade_date=%s rerun=%s",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
rerun,
|
|
)
|
|
if strategy not in self.evaluators:
|
|
raise ValueError(f"selection evaluator is not configured for strategy: {strategy}")
|
|
try:
|
|
source = self.reader.load_execution_source(strategy, target_trade_date)
|
|
run = self.store.prepare_run(
|
|
strategy,
|
|
target_trade_date,
|
|
source,
|
|
rerun=rerun,
|
|
)
|
|
except Exception as exc: # noqa: BLE001 - log the safe prepare boundary and preserve type
|
|
logger.warning(
|
|
"selection_run_prepare_failed strategy=%s target_trade_date=%s "
|
|
"status=failed error_type=%s reason=%s",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
exc.__class__.__name__,
|
|
_safe_item_error(exc),
|
|
)
|
|
raise
|
|
logger.info(
|
|
"selection_run_prepared strategy=%s target_trade_date=%s run_id=%s "
|
|
"market_sync_batch_id=%s target_count=%d eligible_count=%d "
|
|
"coverage=%s status=running",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
run.id,
|
|
source.market_sync_batch_id,
|
|
source.target_count,
|
|
len(source.stocks),
|
|
source.coverage,
|
|
)
|
|
return PreparedSelectionRun(run=run, source=source)
|
|
|
|
def execute(self, prepared: PreparedSelectionRun) -> None:
|
|
"""Evaluate every eligible stock and converge the persisted run status.
|
|
|
|
This method is the boundary used by FastAPI's in-process background
|
|
task. An unexpected batch-level error is recorded before the worker
|
|
returns so the UI never mistakes a lost worker exception for success.
|
|
"""
|
|
|
|
stocks = _unique_stocks(prepared.source.stocks)
|
|
strategy = prepared.run.strategy
|
|
target_trade_date = prepared.source.target_trade_date
|
|
evaluator = self.evaluators[strategy]
|
|
evaluated_count = 0
|
|
selected_stock_count = 0
|
|
signal_count = 0
|
|
failed_count = 0
|
|
missing_turnover_count = 0
|
|
insufficient_history_count = 0
|
|
history_rows = 0
|
|
batch_count = _chunk_count(len(stocks), self.batch_size)
|
|
current_batch = 0
|
|
final_status: SelectionRunStatus = "failed"
|
|
read_seconds = 0.0
|
|
evaluate_seconds = 0.0
|
|
persist_seconds = 0.0
|
|
scoring_seconds = 0.0
|
|
logger.info(
|
|
"selection_run_started strategy=%s target_trade_date=%s run_id=%s "
|
|
"market_sync_batch_id=%s stock_count=%d batch_count=%d worker_count=%d "
|
|
"status=running",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
prepared.run.id,
|
|
prepared.source.market_sync_batch_id,
|
|
len(stocks),
|
|
batch_count,
|
|
self.max_workers,
|
|
)
|
|
try:
|
|
if strategy == "zhixing_b1":
|
|
pattern_cases, pattern_library_error = self._prepare_pattern_cases(prepared.run.id)
|
|
else:
|
|
pattern_cases, pattern_library_error = None, None
|
|
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
|
|
for batch_index, batch_stocks in enumerate(
|
|
_chunks(stocks, self.batch_size),
|
|
start=1,
|
|
):
|
|
current_batch = batch_index
|
|
read_started = time.perf_counter()
|
|
histories = self._load_histories(
|
|
batch_stocks,
|
|
target_trade_date,
|
|
evaluator,
|
|
)
|
|
batch_read_seconds = time.perf_counter() - read_started
|
|
read_seconds += batch_read_seconds
|
|
batch_history_rows = sum(
|
|
len(history.bars) for history in histories if history is not None
|
|
)
|
|
history_rows += batch_history_rows
|
|
batch_missing_turnover = sum(
|
|
not _turnover_present(history, target_trade_date) for history in histories
|
|
)
|
|
logger.info(
|
|
"selection_read_batch_summary strategy=%s target_trade_date=%s "
|
|
"run_id=%s batch=%d batch_count=%d stock_count=%d history_rows=%d "
|
|
"turnover_missing_count=%d status=success read_seconds=%.3f",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
prepared.run.id,
|
|
batch_index,
|
|
batch_count,
|
|
len(batch_stocks),
|
|
batch_history_rows,
|
|
batch_missing_turnover,
|
|
batch_read_seconds,
|
|
)
|
|
|
|
evaluate_started = time.perf_counter()
|
|
evaluations = tuple(
|
|
executor.map(
|
|
self._evaluate_stock,
|
|
batch_stocks,
|
|
histories,
|
|
[target_trade_date] * len(batch_stocks),
|
|
[evaluator] * len(batch_stocks),
|
|
[strategy] * len(batch_stocks),
|
|
[prepared.run.id] * len(batch_stocks),
|
|
[batch_index] * len(batch_stocks),
|
|
)
|
|
)
|
|
batch_evaluate_seconds = time.perf_counter() - evaluate_started
|
|
evaluate_seconds += batch_evaluate_seconds
|
|
status_counts = Counter(evaluation.status for evaluation in evaluations)
|
|
no_signal_reasons = Counter(
|
|
evaluation.reason or "unspecified"
|
|
for evaluation in evaluations
|
|
if evaluation.status == "no_signal"
|
|
)
|
|
missing_turnover_count += status_counts["missing_turnover_rate"]
|
|
insufficient_history_count += status_counts["insufficient_history"]
|
|
for stock, history, evaluation in zip(
|
|
batch_stocks,
|
|
histories,
|
|
evaluations,
|
|
strict=True,
|
|
):
|
|
if evaluation.status not in _FAILURE_STATUSES:
|
|
continue
|
|
logger.warning(
|
|
"selection_item_incomplete strategy=%s target_trade_date=%s "
|
|
"run_id=%s batch=%d ts_code=%s history_rows=%d "
|
|
"turnover_present=%s status=%s error_type=%s reason=%s",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
prepared.run.id,
|
|
batch_index,
|
|
stock.ts_code,
|
|
len(history.bars) if history is not None else 0,
|
|
_turnover_present(history, target_trade_date),
|
|
evaluation.status,
|
|
evaluation.status,
|
|
evaluation.reason or evaluation.status,
|
|
)
|
|
|
|
scoring_started = time.perf_counter()
|
|
items = tuple(
|
|
_to_item(
|
|
stock.ts_code,
|
|
stock.name,
|
|
evaluation,
|
|
pattern_score=self._score_stock(
|
|
strategy,
|
|
prepared.run.id,
|
|
stock,
|
|
history,
|
|
evaluation,
|
|
pattern_cases,
|
|
pattern_library_error,
|
|
),
|
|
)
|
|
for stock, history, evaluation in zip(
|
|
batch_stocks,
|
|
histories,
|
|
evaluations,
|
|
strict=True,
|
|
)
|
|
)
|
|
scoring_seconds += time.perf_counter() - scoring_started
|
|
|
|
evaluated_count += len(items)
|
|
selected_stock_count += sum(item.status == "selected" for item in items)
|
|
signal_count += sum(item.signal_count for item in items)
|
|
failed_count += sum(item.status in _FAILURE_STATUSES for item in items)
|
|
|
|
logger.info(
|
|
"selection_evaluate_batch_summary strategy=%s target_trade_date=%s "
|
|
"run_id=%s batch=%d batch_count=%d stock_count=%d selected_count=%d "
|
|
"no_signal_count=%d insufficient_history_count=%d "
|
|
"missing_target_bar_count=%d missing_turnover_count=%d "
|
|
"data_error_count=%d no_signal_reasons=%s status=complete "
|
|
"evaluate_seconds=%.3f",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
prepared.run.id,
|
|
batch_index,
|
|
batch_count,
|
|
len(items),
|
|
status_counts["selected"],
|
|
status_counts["no_signal"],
|
|
status_counts["insufficient_history"],
|
|
status_counts["missing_target_bar"],
|
|
status_counts["missing_turnover_rate"],
|
|
status_counts["data_error"],
|
|
dict(no_signal_reasons),
|
|
batch_evaluate_seconds,
|
|
)
|
|
|
|
persist_started = time.perf_counter()
|
|
self._record_items(prepared.run.id, items)
|
|
batch_persist_seconds = time.perf_counter() - persist_started
|
|
persist_seconds += batch_persist_seconds
|
|
logger.info(
|
|
"selection_persist_batch_summary strategy=%s target_trade_date=%s "
|
|
"run_id=%s batch=%d batch_count=%d item_count=%d signal_count=%d "
|
|
"status=success persist_seconds=%.3f",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
prepared.run.id,
|
|
batch_index,
|
|
batch_count,
|
|
len(items),
|
|
sum(item.signal_count for item in items),
|
|
batch_persist_seconds,
|
|
)
|
|
|
|
final_status = _run_status(evaluated_count, failed_count)
|
|
self.store.finish_run(
|
|
prepared.run.id,
|
|
final_status,
|
|
evaluated_count=evaluated_count,
|
|
selected_stock_count=selected_stock_count,
|
|
signal_count=signal_count,
|
|
failed_count=failed_count,
|
|
)
|
|
logger.info(
|
|
"selection_run_converged strategy=%s target_trade_date=%s run_id=%s "
|
|
"batch=%d evaluated_count=%d selected_stock_count=%d signal_count=%d "
|
|
"failed_count=%d status=%s",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
prepared.run.id,
|
|
current_batch,
|
|
evaluated_count,
|
|
selected_stock_count,
|
|
signal_count,
|
|
failed_count,
|
|
final_status,
|
|
)
|
|
except Exception as exc: # noqa: BLE001 - worker boundary must persist failure state
|
|
final_status = "failed"
|
|
logger.error(
|
|
"selection_run_failed strategy=%s target_trade_date=%s run_id=%s "
|
|
"batch=%d status=failed error_type=%s reason=%s",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
prepared.run.id,
|
|
current_batch,
|
|
exc.__class__.__name__,
|
|
_safe_item_error(exc),
|
|
)
|
|
try:
|
|
self.store.finish_run(
|
|
prepared.run.id,
|
|
"failed",
|
|
evaluated_count=evaluated_count,
|
|
selected_stock_count=selected_stock_count,
|
|
signal_count=signal_count,
|
|
failed_count=max(failed_count, 1),
|
|
error_type="batch_error",
|
|
error_message=_safe_item_error(exc),
|
|
)
|
|
except Exception: # noqa: BLE001 - preserve the original worker failure
|
|
logger.error(
|
|
"selection_run_failure_persist_failed strategy=%s target_trade_date=%s "
|
|
"run_id=%s batch=%d status=failed error_type=finish_run_failed",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
prepared.run.id,
|
|
current_batch,
|
|
)
|
|
finally:
|
|
logger.info(
|
|
"selection_run_summary strategy=%s target_trade_date=%s run_id=%s "
|
|
"market_sync_batch_id=%s stock_count=%d history_rows=%d batch_count=%d "
|
|
"last_batch=%d worker_count=%d evaluated_count=%d selected_stock_count=%d "
|
|
"signal_count=%d failed_count=%d insufficient_history_count=%d "
|
|
"missing_turnover_count=%d status=%s read_seconds=%.3f "
|
|
"evaluate_seconds=%.3f scoring_seconds=%.3f persist_seconds=%.3f",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
prepared.run.id,
|
|
prepared.source.market_sync_batch_id,
|
|
len(stocks),
|
|
history_rows,
|
|
batch_count,
|
|
current_batch,
|
|
self.max_workers,
|
|
evaluated_count,
|
|
selected_stock_count,
|
|
signal_count,
|
|
failed_count,
|
|
insufficient_history_count,
|
|
missing_turnover_count,
|
|
final_status,
|
|
read_seconds,
|
|
evaluate_seconds,
|
|
scoring_seconds,
|
|
persist_seconds,
|
|
)
|
|
|
|
def _prepare_pattern_cases(
|
|
self,
|
|
run_id: str,
|
|
) -> tuple[tuple[PatternCase, ...] | None, str | None]:
|
|
"""Load the complete case library once without failing selection."""
|
|
|
|
if not self.pattern_scoring_enabled:
|
|
return None, None
|
|
if self.pattern_case_loader is None or self.pattern_scorer is None:
|
|
reason = "pattern scoring is enabled but not configured"
|
|
logger.error("selection_pattern_library_failed run_id=%s reason=%s", run_id, reason)
|
|
return None, reason
|
|
try:
|
|
return self.pattern_case_loader.load(), None
|
|
except Exception as exc: # noqa: BLE001 - scoring enrichment must not fail selection
|
|
reason = _safe_item_error(exc)
|
|
logger.warning(
|
|
"selection_pattern_library_failed run_id=%s error_type=%s reason=%s",
|
|
run_id,
|
|
exc.__class__.__name__,
|
|
reason,
|
|
)
|
|
return None, reason
|
|
|
|
def _score_stock(
|
|
self,
|
|
strategy: StrategyName,
|
|
run_id: str,
|
|
stock: SelectionStock,
|
|
history: StockHistory | None,
|
|
evaluation: SelectionEvaluation,
|
|
cases: tuple[PatternCase, ...] | None,
|
|
library_error: str | None,
|
|
) -> PatternScore:
|
|
"""Score one selected stock once and isolate enrichment failures."""
|
|
|
|
if (
|
|
strategy != "zhixing_b1"
|
|
or not self.pattern_scoring_enabled
|
|
or evaluation.status != "selected"
|
|
):
|
|
return PatternScore()
|
|
if library_error is not None:
|
|
return PatternScore.failed(library_error)
|
|
if history is None or cases is None or self.pattern_scorer is None:
|
|
return PatternScore.failed("pattern scoring history or case library is unavailable")
|
|
try:
|
|
return self.pattern_scorer.score(history, cases)
|
|
except Exception as exc: # noqa: BLE001 - one score must not fail the selection run
|
|
reason = _safe_item_error(exc)
|
|
logger.warning(
|
|
"selection_pattern_score_failed run_id=%s ts_code=%s error_type=%s reason=%s",
|
|
run_id,
|
|
stock.ts_code,
|
|
exc.__class__.__name__,
|
|
reason,
|
|
)
|
|
return PatternScore.failed(reason)
|
|
|
|
def _load_histories(
|
|
self,
|
|
stocks: Sequence[SelectionStock],
|
|
target_trade_date: date,
|
|
evaluator: SelectionEvaluator,
|
|
) -> tuple[StockHistory | None, ...]:
|
|
"""Load one chunk when the reader supports it, with old-path fallback."""
|
|
|
|
typed_stocks = tuple(stocks)
|
|
loader = getattr(self.reader, "load_histories", None)
|
|
if callable(loader):
|
|
batch_reader = cast(BatchSelectionUniverseReader, self.reader)
|
|
loaded = batch_reader.load_histories(typed_stocks, target_trade_date)
|
|
histories_by_code = {history.ts_code: history for history in loaded}
|
|
return tuple(
|
|
histories_by_code.get(
|
|
stock.ts_code,
|
|
StockHistory(ts_code=stock.ts_code, name=stock.name),
|
|
)
|
|
for stock in typed_stocks
|
|
)
|
|
|
|
execute_history = getattr(evaluator, "execute_history", None)
|
|
if callable(execute_history):
|
|
return tuple(
|
|
self.reader.load_history(stock.ts_code, target_trade_date) for stock in typed_stocks
|
|
)
|
|
return (None,) * len(typed_stocks)
|
|
|
|
def _evaluate_stock(
|
|
self,
|
|
stock: SelectionStock,
|
|
history: StockHistory | None,
|
|
target_trade_date: date,
|
|
evaluator: SelectionEvaluator,
|
|
strategy: StrategyName,
|
|
run_id: str,
|
|
batch_index: int,
|
|
) -> SelectionEvaluation:
|
|
"""Evaluate one stock inside a worker and isolate its exception."""
|
|
|
|
ts_code = stock.ts_code
|
|
try:
|
|
execute_history: Callable[[StockHistory, date], SelectionEvaluation] | None = getattr(
|
|
evaluator,
|
|
"execute_history",
|
|
None,
|
|
)
|
|
if history is not None and execute_history is not None:
|
|
return execute_history(history, target_trade_date)
|
|
return evaluator.execute(ts_code, target_trade_date)
|
|
except Exception as exc: # noqa: BLE001 - isolate one stock from the batch
|
|
logger.warning(
|
|
"selection_item_failed strategy=%s target_trade_date=%s run_id=%s "
|
|
"batch=%d ts_code=%s history_rows=%d turnover_present=%s "
|
|
"status=data_error error_type=%s reason=%s",
|
|
strategy,
|
|
target_trade_date.isoformat(),
|
|
run_id,
|
|
batch_index,
|
|
ts_code,
|
|
len(history.bars) if history is not None else 0,
|
|
_turnover_present(history, target_trade_date),
|
|
exc.__class__.__name__,
|
|
_safe_item_error(exc),
|
|
)
|
|
return SelectionEvaluation(
|
|
ts_code=ts_code,
|
|
target_trade_date=target_trade_date,
|
|
status="data_error",
|
|
reason=_safe_item_error(exc),
|
|
)
|
|
|
|
def _record_items(self, run_id: str, items: Sequence[SelectionRunItem]) -> None:
|
|
"""Use batch persistence while retaining the old single-item seam."""
|
|
|
|
record_items = getattr(self.store, "record_items", None)
|
|
if callable(record_items):
|
|
batch_store = cast(BatchSelectionRunStore, self.store)
|
|
batch_store.record_items(run_id, tuple(items))
|
|
return
|
|
for item in items:
|
|
self.store.record_item(run_id, item)
|
|
|
|
def get_run(
|
|
self,
|
|
run_id: str,
|
|
*,
|
|
query: SelectionResultQuery | None = None,
|
|
) -> SelectionRun | None:
|
|
"""Read one persisted run for polling with optional sector filtering."""
|
|
|
|
effective = query or SelectionResultQuery()
|
|
if not effective.sector:
|
|
return self.store.get_run(run_id, query=query)
|
|
identity = self.store.get_run_identity(run_id)
|
|
if identity is None:
|
|
return None
|
|
member_codes = self._sector_member_codes(identity.target_trade_date, effective.sector)
|
|
if member_codes is None:
|
|
return self.store.get_run(run_id, query=query)
|
|
return self.store.get_run(
|
|
run_id,
|
|
query=replace(effective, sector=None),
|
|
sector_stock_codes=member_codes,
|
|
)
|
|
|
|
def get_latest(
|
|
self,
|
|
strategy: StrategyName,
|
|
target_trade_date: date | None = None,
|
|
*,
|
|
query: SelectionResultQuery | None = None,
|
|
) -> SelectionRun | None:
|
|
"""Read the current result by date or the latest result for a strategy."""
|
|
|
|
effective = query or SelectionResultQuery()
|
|
if not effective.sector:
|
|
return self.store.get_latest_run(strategy, target_trade_date, query=query)
|
|
identity = self.store.get_latest_run_identity(strategy, target_trade_date)
|
|
if identity is None:
|
|
return None
|
|
member_codes = self._sector_member_codes(identity.target_trade_date, effective.sector)
|
|
if member_codes is None:
|
|
return self.store.get_latest_run(strategy, target_trade_date, query=query)
|
|
return self.store.get_latest_run(
|
|
strategy,
|
|
identity.target_trade_date,
|
|
query=replace(effective, sector=None),
|
|
sector_stock_codes=member_codes,
|
|
)
|
|
|
|
def _sector_member_codes(
|
|
self,
|
|
target_trade_date: date,
|
|
sector_code: str,
|
|
) -> tuple[str, ...] | None:
|
|
"""Resolve one sub-industry's members, or None when the port is absent."""
|
|
|
|
if self.sector_reader is None:
|
|
return None
|
|
return self.sector_reader.sector_member_codes(
|
|
target_trade_date,
|
|
sector_code,
|
|
sector_type="industry",
|
|
)
|
|
|
|
def list_sector_counts(
|
|
self,
|
|
strategy: StrategyName,
|
|
target_trade_date: date | None = None,
|
|
) -> SelectionSectorAggregates | None:
|
|
"""Aggregate the current run's selected stocks by sub-industry."""
|
|
|
|
run = self.store.get_latest_run(strategy, target_trade_date)
|
|
if run is None:
|
|
return None
|
|
selected_codes = [
|
|
item.ts_code
|
|
for item in run.items
|
|
if item.status == "selected" and item.signal_count > 0
|
|
]
|
|
membership = self._sector_membership(selected_codes, run.target_trade_date)
|
|
return SelectionSectorAggregates(
|
|
run=run,
|
|
snapshot_trade_date=membership.snapshot_trade_date,
|
|
sector_type="industry",
|
|
sectors=membership.sector_counts,
|
|
)
|
|
|
|
def _sector_membership(
|
|
self,
|
|
stock_codes: Sequence[str],
|
|
target_trade_date: date,
|
|
) -> SelectionSectorMembership:
|
|
"""Read sub-industry counts for a stock set, tolerating a missing port."""
|
|
|
|
if self.sector_reader is None or not stock_codes:
|
|
return SelectionSectorMembership(snapshot_trade_date=None, sector_counts=())
|
|
return self.sector_reader.sector_counts(
|
|
stock_codes,
|
|
target_trade_date,
|
|
sector_type="industry",
|
|
)
|
|
|
|
|
|
def _to_item(
|
|
ts_code: str,
|
|
name: str,
|
|
evaluation: SelectionEvaluation,
|
|
*,
|
|
pattern_score: PatternScore | None = None,
|
|
) -> SelectionRunItem:
|
|
"""Translate a single-stock domain result into a stored item."""
|
|
|
|
return SelectionRunItem(
|
|
ts_code=ts_code,
|
|
name=name or (evaluation.signals[0].name if evaluation.signals else ""),
|
|
status=evaluation.status,
|
|
signal_count=len(evaluation.signals),
|
|
reason=evaluation.reason,
|
|
pattern_score=pattern_score or PatternScore(),
|
|
signals=evaluation.signals,
|
|
)
|
|
|
|
|
|
def _run_status(evaluated_count: int, failed_count: int) -> SelectionRunStatus:
|
|
"""Map per-stock outcomes into a visible batch status."""
|
|
|
|
if failed_count == 0:
|
|
return "success"
|
|
if evaluated_count == 0 or failed_count >= evaluated_count:
|
|
return "failed"
|
|
return "partial_success"
|
|
|
|
|
|
def _safe_item_error(error: Exception) -> str:
|
|
"""Keep per-stock failure context readable without persisting tracebacks."""
|
|
|
|
return " ".join(str(error).split())[:500] or error.__class__.__name__
|
|
|
|
|
|
def _turnover_present(history: StockHistory | None, target_trade_date: date) -> bool:
|
|
"""Return whether target-day Tushare turnover is available for diagnostics."""
|
|
|
|
if history is None:
|
|
return False
|
|
basic = history.daily_basic.get(target_trade_date)
|
|
return basic is not None and basic.turnover_rate is not None
|
|
|
|
|
|
def _chunks(
|
|
values: Sequence[SelectionStock],
|
|
size: int,
|
|
) -> tuple[tuple[SelectionStock, ...], ...]:
|
|
"""Split a stable stock sequence into bounded immutable chunks."""
|
|
|
|
return tuple(tuple(values[index : index + size]) for index in range(0, len(values), size))
|
|
|
|
|
|
def _chunk_count(value_count: int, size: int) -> int:
|
|
"""Return the number of chunks without materializing empty chunks."""
|
|
|
|
return (value_count + size - 1) // size
|
|
|
|
|
|
def _unique_stocks(stocks: Sequence[SelectionStock]) -> tuple[SelectionStock, ...]:
|
|
"""Keep the first source row for each stock so it is evaluated once."""
|
|
|
|
seen: set[str] = set()
|
|
unique: list[SelectionStock] = []
|
|
for stock in stocks:
|
|
ts_code = stock.ts_code
|
|
if ts_code in seen:
|
|
continue
|
|
seen.add(ts_code)
|
|
unique.append(stock)
|
|
return tuple(unique)
|
|
|
|
|
|
__all__ = [
|
|
"PreparedSelectionRun",
|
|
"RunZhixingB1",
|
|
"SelectionRerunRequired",
|
|
"SelectionRunInProgress",
|
|
"SelectionSectorAggregates",
|
|
]
|