"""Read-only application use case for one selected stock's chart series.""" from __future__ import annotations from dataclasses import dataclass from datetime import date from math import isfinite from typing import Literal, cast import pandas as pd from ..domain.indicators import compute_kdj, compute_zhixing_lines from ..domain.ports import MarketDataReader SELECTION_CHART_LIMIT = 250 class SelectionChartNotFound(LookupError): """No qfq daily history exists through the requested target date.""" @dataclass(frozen=True, slots=True) class SelectionChartPoint: """One date-aligned OHLCV, KDJ, and Zhixing-line point.""" trade_date: date open: float | None high: float | None low: float | None close: float | None volume: float | None k: float | None d: float | None j: float | None trend_white: float | None trend_yellow: float | None @dataclass(frozen=True, slots=True) class SelectionChart: """A bounded ascending qfq chart series for one stock and target date.""" ts_code: str name: str target_trade_date: date source_adj: Literal["qfq"] points: tuple[SelectionChartPoint, ...] class GetSelectionChart: """Load one qfq history, compute KDJ and Zhixing lines, then bound the series.""" def __init__(self, reader: MarketDataReader) -> None: """Inject the market-data reader owned by the selection context.""" self.reader = reader def execute(self, ts_code: str, target_trade_date: date) -> SelectionChart: """Return at most 250 points without changing full-history indicator state. Args: ts_code: Tushare stock identifier selected by the user. target_trade_date: Inclusive historical boundary for the chart. Returns: An ascending qfq chart series aligned by trade date. Raises: SelectionChartNotFound: If no qfq bars exist through the target date. MarketDataReaderError: If the injected reader cannot complete the read. """ history = self.reader.load_history(ts_code, target_trade_date) bars = tuple(bar for bar in history.bars if bar.trade_date <= target_trade_date) if not bars: raise SelectionChartNotFound( f"chart data not found for {ts_code} at {target_trade_date.isoformat()}" ) frame = pd.DataFrame( { "low": [bar.low for bar in bars], "high": [bar.high for bar in bars], "close": [bar.close for bar in bars], } ) kdj = compute_kdj(frame) white, yellow = compute_zhixing_lines(frame["close"]) start = max(0, len(bars) - SELECTION_CHART_LIMIT) points = tuple( SelectionChartPoint( trade_date=bar.trade_date, open=bar.open, high=bar.high, low=bar.low, close=bar.close, volume=bar.volume, k=_finite_or_none(kdj.iloc[index]["K"]), d=_finite_or_none(kdj.iloc[index]["D"]), j=_finite_or_none(kdj.iloc[index]["J"]), trend_white=_finite_or_none(white.iloc[index]), trend_yellow=_finite_or_none(yellow.iloc[index]), ) for index, bar in enumerate(bars) if index >= start ) return SelectionChart( ts_code=history.ts_code, name=history.name, target_trade_date=target_trade_date, source_adj="qfq", points=points, ) def _finite_or_none(value: object) -> float | None: """Convert one Pandas scalar to a finite JSON-safe float or ``None``.""" try: number = float(cast(float, value)) except (TypeError, ValueError): return None return number if isfinite(number) else None __all__ = [ "GetSelectionChart", "SELECTION_CHART_LIMIT", "SelectionChart", "SelectionChartNotFound", "SelectionChartPoint", ]