zhixing-system/zhixing-server/src/zhixing_server/modules/selection/domain/runs.py

"""Domain contracts for persisted historical selection runs."""

from __future__ import annotations

from collections.abc import Sequence
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from typing import Literal, Protocol

from .models import SelectionEvaluationStatus, SelectionSignal, StockHistory
from .pattern_scoring import PatternScore

SelectionRunStatus = Literal["running", "success", "partial_success", "failed"]
SelectionRunItemStatus = SelectionEvaluationStatus
SelectionSignalCategoryFilter = Literal["pullback", "oversold", "original"]
SelectionResultSort = Literal["code", "score_desc", "score_asc"]


@dataclass(frozen=True, slots=True)
class SelectionResultQuery:
    """Validated query options for a paged selection-result read."""

    page: int = 1
    page_size: int = 10
    search: str | None = None
    category: SelectionSignalCategoryFilter | None = None
    sort: SelectionResultSort = "code"


@dataclass(frozen=True, slots=True)
class SelectionStock:
    """One eligible current stock that will be evaluated for a run."""

    ts_code: str
    name: str


@dataclass(frozen=True, slots=True)
class SelectionExecutionSource:
    """Market-data batch and eligible stock snapshot used by one run."""

    market_sync_batch_id: str
    target_trade_date: date
    target_count: int
    valid_count: int
    coverage: Decimal
    stocks: tuple[SelectionStock, ...] = field(default_factory=tuple)


@dataclass(frozen=True, slots=True)
class SelectionRunItem:
    """Persistable per-stock evaluation state and its independent signals."""

    ts_code: str
    name: str
    status: SelectionRunItemStatus
    signal_count: int = 0
    reason: str | None = None
    pattern_score: PatternScore = field(default_factory=PatternScore)
    signals: tuple[SelectionSignal, ...] = field(default_factory=tuple)


@dataclass(frozen=True, slots=True)
class SelectionRun:
    """A current execution attempt and its materialized result rows."""

    id: str
    strategy: Literal["zhixing_b1"]
    target_trade_date: date
    market_sync_batch_id: str | None
    status: SelectionRunStatus
    target_count: int
    eligible_count: int
    evaluated_count: int
    selected_stock_count: int
    signal_count: int
    failed_count: int
    coverage: Decimal
    error_type: str | None = None
    error_message: str | None = None
    created_at: datetime | None = None
    finished_at: datetime | None = None
    items: tuple[SelectionRunItem, ...] = field(default_factory=tuple)
    signals: tuple[SelectionSignal, ...] = field(default_factory=tuple)
    stocks_total: int | None = None


class SelectionRunError(RuntimeError):
    """Base class for expected selection-run persistence failures."""


class SelectionRunInProgress(SelectionRunError):
    """The requested strategy and date already have a running attempt."""


class SelectionRerunRequired(SelectionRunError):
    """A terminal result exists and an explicit rerun confirmation is missing."""


class SelectionRunStoreError(SelectionRunError):
    """The selection-run repository could not complete a database operation."""


class SelectionRunStore(Protocol):
    """Persistence port for current selection runs and their materialized rows."""

    def prepare_run(
        self,
        strategy: Literal["zhixing_b1"],
        target_trade_date: date,
        source: SelectionExecutionSource,
        *,
        rerun: bool,
    ) -> SelectionRun: ...

    def record_item(self, run_id: str, item: SelectionRunItem) -> None: ...

    def finish_run(
        self,
        run_id: str,
        status: SelectionRunStatus,
        *,
        evaluated_count: int,
        selected_stock_count: int,
        signal_count: int,
        failed_count: int,
        error_type: str | None = None,
        error_message: str | None = None,
    ) -> None: ...

    def get_run(
        self,
        run_id: str,
        *,
        query: SelectionResultQuery | None = None,
    ) -> SelectionRun | None: ...

    def get_latest_run(
        self,
        strategy: Literal["zhixing_b1"],
        target_trade_date: date | None = None,
        *,
        query: SelectionResultQuery | None = None,
    ) -> SelectionRun | None: ...


class SelectionUniverseReader(Protocol):
    """Read a qualified market-data source snapshot for one strategy run."""

    def load_execution_source(
        self,
        strategy: str,
        target_trade_date: date,
    ) -> SelectionExecutionSource: ...

    def load_history(self, ts_code: str, target_trade_date: date) -> StockHistory: ...


class BatchSelectionRunStore(Protocol):
    """Optional batch-write extension for selection stores."""

    def record_items(self, run_id: str, items: Sequence[SelectionRunItem]) -> None: ...


class BatchSelectionUniverseReader(Protocol):
    """Optional bounded batch-history extension for selection readers."""

    def load_histories(
        self,
        stocks: Sequence[SelectionStock],
        target_trade_date: date,
    ) -> tuple[StockHistory, ...]: ...