my-home/zhixing-system
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(selection): 新增个股技术图表接口

Browse Source
This commit is contained in:
yuxuanhui
2026-09-01 13:37:11 +08:00
parent c8b08add76
commit 9145fe4b2c
4 changed files with 438 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zhixing-server/src/zhixing_server/modules/selection/application/chart.py
+127
View File
@@ -0,0 +1,127 @@
"""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",
]
zhixing-server/src/zhixing_server/modules/selection/presentation/http.py
+86 -7
View File
@@ -9,6 +9,11 @@ from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, s
from pydantic import BaseModel, Field
from zhixing_server.bootstrap.config import Settings, get_settings
from zhixing_server.modules.selection.application.chart import (
GetSelectionChart,
SelectionChart,
SelectionChartNotFound,
)
from zhixing_server.modules.selection.application.run import (
RunZhixingB1,
)
@@ -17,6 +22,7 @@ from zhixing_server.modules.selection.domain.pattern_scoring import (
PatternScore,
ZhixingB1PatternScorer,
)
from zhixing_server.modules.selection.domain.ports import MarketDataReaderError
from zhixing_server.modules.selection.domain.runs import (
SelectionRerunRequired,
SelectionResultQuery,
@@ -140,6 +146,30 @@ class SelectionStockResponse(BaseModel):
signals: list[SelectionSignalResponse] = Field(default_factory=_empty_signals)
class SelectionChartPointResponse(BaseModel):
"""One date-aligned qfq OHLCV and KDJ 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
class SelectionChartResponse(BaseModel):
"""A bounded ascending chart series for one selected stock."""
ts_code: str
name: str
target_trade_date: date
source_adj: Literal["qfq"]
points: list[SelectionChartPointResponse]
def _empty_stocks() -> list[SelectionStockResponse]:
"""Create a typed default stock-result list."""
@@ -192,6 +222,15 @@ def get_selection_service(
)
def get_selection_chart_service(
settings: Annotated[Settings, Depends(get_settings)],
) -> GetSelectionChart:
"""Build the read-only chart use case on the shared selection pool."""
pool = get_selection_postgres_pool(settings)
return GetSelectionChart(PostgresMarketDataReader(settings, pool=pool))
def get_selection_postgres_pool(settings: Settings) -> SelectionPostgresPool:
"""Return the cached bounded pool shared by selection adapters."""
@@ -220,6 +259,26 @@ def _close_cached_selection_pools() -> None:
atexit.register(_close_cached_selection_pools)
@selection_router.get(
"/stocks/{ts_code}/chart",
response_model=SelectionChartResponse,
)
def get_selection_chart(
ts_code: str,
target_trade_date: date,
service: Annotated[GetSelectionChart, Depends(get_selection_chart_service)],
) -> SelectionChartResponse:
"""Return one stock's bounded qfq OHLCV and KDJ history."""
try:
chart = service.execute(ts_code, target_trade_date)
except SelectionChartNotFound as exc:
raise _http_error(404, "chart_data_not_found", str(exc)) from exc
except MarketDataReaderError as exc:
raise _http_error(503, "selection_storage_unavailable", str(exc)) from exc
return _chart_response(chart)
@selection_router.post(
"/runs",
response_model=SelectionRunAcceptedResponse,
@@ -387,13 +446,6 @@ def _pattern_score_response(score: PatternScore) -> SelectionPatternScoreRespons
return None
if score.status == "failed":
return SelectionPatternScoreResponse(status="failed", reason=score.reason)
if score.status == "below_threshold":
return SelectionPatternScoreResponse(
status="below_threshold",
threshold=score.threshold,
version=score.version,
reason="未匹配到评分阈值以上案例",
)
if score.case is None or score.breakdown is None:
return SelectionPatternScoreResponse(status="failed", reason="评分结果不完整")
return SelectionPatternScoreResponse(
@@ -411,6 +463,31 @@ def _pattern_score_response(score: PatternScore) -> SelectionPatternScoreRespons
)
def _chart_response(chart: SelectionChart) -> SelectionChartResponse:
"""Translate an application chart without leaking domain dataclasses."""
return SelectionChartResponse(
ts_code=chart.ts_code,
name=chart.name,
target_trade_date=chart.target_trade_date,
source_adj=chart.source_adj,
points=[
SelectionChartPointResponse(
trade_date=point.trade_date,
open=point.open,
high=point.high,
low=point.low,
close=point.close,
volume=point.volume,
k=point.k,
d=point.d,
j=point.j,
)
for point in chart.points
],
)
def _result_query(
page: int,
page_size: int,
@@ -440,10 +517,12 @@ def _http_error(code: int, error_type: str, message: str) -> HTTPException:
__all__ = [
"SelectionChartResponse",
"SelectionResultsResponse",
"SelectionRunAcceptedResponse",
"SelectionRunRequest",
"SelectionStockResponse",
"get_selection_chart_service",
"get_selection_service",
"get_selection_postgres_pool",
"selection_router",
zhixing-server/tests/test_selection_http.py
+124 -5
View File
@@ -10,6 +10,11 @@ from fastapi.testclient import TestClient
import zhixing_server.modules.selection.presentation.http as selection_http
from zhixing_server.bootstrap.app import create_app
from zhixing_server.bootstrap.config import Settings
from zhixing_server.modules.selection.application.chart import (
SelectionChart,
SelectionChartNotFound,
SelectionChartPoint,
)
from zhixing_server.modules.selection.application.run import PreparedSelectionRun
from zhixing_server.modules.selection.domain.models import SelectionSignal, ZhixingB1Category
from zhixing_server.modules.selection.domain.pattern_scoring import (
@@ -18,6 +23,7 @@ from zhixing_server.modules.selection.domain.pattern_scoring import (
PatternScore,
PatternScoreBreakdown,
)
from zhixing_server.modules.selection.domain.ports import MarketDataReaderError
from zhixing_server.modules.selection.domain.runs import (
SelectionExecutionSource,
SelectionRerunRequired,
@@ -30,6 +36,7 @@ from zhixing_server.modules.selection.infrastructure.postgres_reader import (
SelectionMarketDataNotReady,
)
from zhixing_server.modules.selection.presentation.http import (
get_selection_chart_service,
get_selection_postgres_pool,
get_selection_service,
)
@@ -97,6 +104,40 @@ class FakeSelectionService:
return self.run
class FakeChartService:
"""Return or fail one deterministic chart response."""
def __init__(self) -> None:
self.mode = "ok"
self.request: tuple[str, date] | None = None
def execute(self, ts_code: str, target_trade_date: date) -> SelectionChart:
self.request = (ts_code, target_trade_date)
if self.mode == "missing":
raise SelectionChartNotFound("chart unavailable")
if self.mode == "storage_error":
raise MarketDataReaderError("chart storage unavailable")
return SelectionChart(
ts_code=ts_code,
name="平安银行",
target_trade_date=target_trade_date,
source_adj="qfq",
points=(
SelectionChartPoint(
trade_date=target_trade_date,
open=10.0,
high=11.0,
low=9.5,
close=10.5,
volume=1000.0,
k=52.0,
d=48.0,
j=60.0,
),
),
)
def _run(run_id: str, status: str) -> SelectionRun:
original_signal = SelectionSignal(
ts_code="000001.SZ",
@@ -153,9 +194,14 @@ def _run(run_id: str, status: str) -> SelectionRun:
)
def _client(service: FakeSelectionService) -> TestClient:
def _client(
service: FakeSelectionService,
chart_service: FakeChartService | None = None,
) -> TestClient:
app = create_app()
app.dependency_overrides[get_selection_service] = lambda: service
if chart_service is not None:
app.dependency_overrides[get_selection_chart_service] = lambda: chart_service
return TestClient(app)
@@ -295,12 +341,21 @@ def test_query_returns_persisted_signal_details() -> None:
),
{
"status": "below_threshold",
"value": None,
"value": 42.5,
"threshold": 60.0,
"version": PATTERN_SCORING_VERSION,
"case": None,
"breakdown": None,
"reason": "未匹配到评分阈值以上案例",
"case": {
"id": "case_001",
"name": "华纳药厂",
"breakout_date": "2025-05-12",
},
"breakdown": {
"trend_structure": 40.0,
"kdj_state": 42.0,
"volume_pattern": 43.0,
"price_shape": 44.0,
},
"reason": None,
},
),
(
@@ -370,6 +425,70 @@ def test_query_rejects_invalid_page_size() -> None:
assert response.status_code == 422
def test_chart_returns_bounded_qfq_contract() -> None:
chart_service = FakeChartService()
response = _client(FakeSelectionService(), chart_service).get(
"/api/v1/selection/stocks/000001.SZ/chart",
params={"target_trade_date": "2026-08-08"},
)
assert response.status_code == 200
assert chart_service.request == ("000001.SZ", TARGET)
assert response.json() == {
"ts_code": "000001.SZ",
"name": "平安银行",
"target_trade_date": "2026-08-08",
"source_adj": "qfq",
"points": [
{
"trade_date": "2026-08-08",
"open": 10.0,
"high": 11.0,
"low": 9.5,
"close": 10.5,
"volume": 1000.0,
"k": 52.0,
"d": 48.0,
"j": 60.0,
}
],
}
@pytest.mark.parametrize(
("mode", "status_code", "error_code"),
[
("missing", 404, "chart_data_not_found"),
("storage_error", 503, "selection_storage_unavailable"),
],
)
def test_chart_maps_application_errors(
mode: str,
status_code: int,
error_code: str,
) -> None:
chart_service = FakeChartService()
chart_service.mode = mode
response = _client(FakeSelectionService(), chart_service).get(
"/api/v1/selection/stocks/000001.SZ/chart",
params={"target_trade_date": "2026-08-08"},
)
assert response.status_code == status_code
assert response.json()["detail"]["code"] == error_code
def test_chart_requires_a_valid_target_date() -> None:
response = _client(FakeSelectionService(), FakeChartService()).get(
"/api/v1/selection/stocks/000001.SZ/chart",
params={"target_trade_date": "not-a-date"},
)
assert response.status_code == 422
def test_query_forwards_score_ascending_sort() -> None:
service = FakeSelectionService(_run("run-http", "success"))
zhixing-server/tests/unit/selection/test_chart.py
+101
View File
@@ -0,0 +1,101 @@
"""Application tests for bounded qfq chart series."""
from datetime import date, timedelta
import pandas as pd
import pytest
from zhixing_server.modules.selection.application.chart import (
GetSelectionChart,
SelectionChartNotFound,
)
from zhixing_server.modules.selection.domain.indicators import compute_kdj
from zhixing_server.modules.selection.domain.models import SelectionBar, StockHistory
class FakeReader:
"""Return one deterministic history through the requested date."""
def __init__(self, history: StockHistory) -> None:
self.history = history
self.request: tuple[str, date] | None = None
def load_history(self, ts_code: str, target_trade_date: date) -> StockHistory:
self.request = (ts_code, target_trade_date)
return self.history
def _history(count: int) -> StockHistory:
start = date(2025, 1, 1)
bars = tuple(
SelectionBar(
trade_date=start + timedelta(days=index),
open=10.0 + index / 100,
high=11.0 + index / 100,
low=9.0 + index / 100,
close=10.5 + index / 100,
volume=1000.0 + index,
)
for index in range(count)
)
return StockHistory(ts_code="000001.SZ", name="平安银行", bars=bars)
def test_chart_computes_on_full_history_before_returning_last_250_points() -> None:
history = _history(260)
reader = FakeReader(history)
target = history.bars[-1].trade_date
chart = GetSelectionChart(reader).execute(history.ts_code, target)
assert reader.request == ("000001.SZ", target)
assert chart.source_adj == "qfq"
assert len(chart.points) == 250
assert chart.points[0].trade_date == history.bars[10].trade_date
assert chart.points[-1].trade_date == target
frame = pd.DataFrame(
{
"low": [bar.low for bar in history.bars],
"high": [bar.high for bar in history.bars],
"close": [bar.close for bar in history.bars],
}
)
expected = compute_kdj(frame).iloc[-1]
assert chart.points[-1].k == pytest.approx(float(expected["K"]))
assert chart.points[-1].d == pytest.approx(float(expected["D"]))
assert chart.points[-1].j == pytest.approx(float(expected["J"]))
def test_chart_filters_future_rows_and_preserves_nullable_points() -> None:
history = _history(12)
target = history.bars[-2].trade_date
incomplete = SelectionBar(
trade_date=history.bars[3].trade_date,
open=history.bars[3].open,
high=None,
low=history.bars[3].low,
close=None,
volume=None,
)
history = StockHistory(
ts_code=history.ts_code,
name=history.name,
bars=history.bars[:3] + (incomplete,) + history.bars[4:],
)
chart = GetSelectionChart(FakeReader(history)).execute(history.ts_code, target)
assert chart.points[-1].trade_date == target
assert all(point.trade_date <= target for point in chart.points)
assert chart.points[3].high is None
assert chart.points[3].volume is None
assert chart.points[3].k is None
assert chart.points[3].d is None
assert chart.points[3].j is None
def test_chart_rejects_empty_history() -> None:
reader = FakeReader(StockHistory(ts_code="000001.SZ", name="平安银行"))
with pytest.raises(SelectionChartNotFound, match="chart data not found"):
GetSelectionChart(reader).execute("000001.SZ", date(2026, 8, 8))