128 lines
3.7 KiB
Python
128 lines
3.7 KiB
Python
|
"""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
|
||
|
|
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 and KDJ point in the public chart vocabulary."""
|
||
|
|
|
||
|
|
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
|
||
|
|
|
||
|
|
|
||
|
|
@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, then bound the response 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 KDJ's full-history 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)
|
||
|
|
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"]),
|
||
|
|
)
|
||
|
|
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",
|
||
|
|
]
|