feat(home): 添加市场数据概览

This commit is contained in:
yuxuanhui
2026-08-07 13:29:31 +08:00
parent 9023e00213
commit b0e846d79c
33 changed files with 2229 additions and 18 deletions
@@ -3,8 +3,10 @@
from fastapi import APIRouter
from zhixing_server.interfaces.http.system import operational_router, system_router
from zhixing_server.modules.market_data.presentation.home import home_router
api_v1_router = APIRouter(prefix="/api/v1")
api_v1_router.include_router(system_router, prefix="/system", tags=["system"])
api_v1_router.include_router(home_router, prefix="/home", tags=["home"])
__all__ = ["api_v1_router", "operational_router"]
@@ -0,0 +1,27 @@
"""Application port and use case for the Home market-data overview."""
from typing import Protocol
from ..domain.overview import MarketDataOverview
class MarketDataOverviewReader(Protocol):
"""Read port implemented by the PostgreSQL market-data adapter."""
def read_overview(self) -> MarketDataOverview:
"""Return the current Home overview without changing market data."""
...
def get_market_data_overview(reader: MarketDataOverviewReader) -> MarketDataOverview:
"""Execute the overview read use case at the application boundary.
Args:
reader: A bounded-context read adapter supplied by application wiring.
Returns:
The current market-data overview read model.
"""
return reader.read_overview()
@@ -0,0 +1,56 @@
"""Read models for the market-data overview shown on Home."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from typing import Literal
OverviewStatus = Literal["updating", "success", "partial_success", "failed", "no_data"]
LatestUpdateStatus = Literal["updating", "success", "partial_success", "failed"]
@dataclass(frozen=True, slots=True)
class OverviewFailure:
"""A target stock that lacks a complete set of facts for one trading day."""
ts_code: str
name: str
failure_type: str
reason: str
trade_date: date
@dataclass(frozen=True, slots=True)
class OverviewBatchError:
"""An error recorded for a batch rather than for one stock."""
item_key: str
error_type: str
reason: str
@dataclass(frozen=True, slots=True)
class MarketDataUpdate:
"""The selected synchronization result and its stock-level facts."""
trade_date: date
completed_at: datetime | None
status: LatestUpdateStatus
target_count: int
successful_count: int
failed_count: int
valid_count: int
coverage_rate: Decimal
failures: tuple[OverviewFailure, ...]
batch_errors: tuple[OverviewBatchError, ...]
@dataclass(frozen=True, slots=True)
class MarketDataOverview:
"""Read-only Home payload before transport-specific serialization."""
status: OverviewStatus
stock_pool_count: int
latest_update: MarketDataUpdate | None
@@ -12,6 +12,13 @@ from uuid import uuid4
import psycopg
from ..domain.models import Bar, DailyBasic, Stock, SyncWindow
from ..domain.overview import (
LatestUpdateStatus,
MarketDataOverview,
MarketDataUpdate,
OverviewBatchError,
OverviewFailure,
)
from ..domain.ports import MarketDataRepositoryError, WriteResult
@@ -25,6 +32,209 @@ class PostgresMarketDataRepository:
def __init__(self, database_url: str) -> None:
self.database_url = database_url
def read_overview(self) -> MarketDataOverview:
"""Aggregate the active stock pool and latest synchronization facts.
The latest target date is selected before batch state, so a running
retry for an older date cannot hide a newer completed target. All
facts are read through one connection to keep the overview internally
consistent without introducing a write-side transaction.
Returns:
A domain read model suitable for the Home HTTP presentation layer.
Raises:
MarketDataRepositoryError: If PostgreSQL cannot complete the read.
"""
with self._connection() as connection:
cursor = connection.cursor()
stock_pool_count = int(
cursor.execute(
"SELECT count(*) FROM market_stock WHERE is_active = true"
).fetchone()[0]
)
batch = cursor.execute(
"""
WITH latest_date AS (
SELECT max(target_trade_date) AS target_trade_date
FROM market_sync_batch
)
SELECT id, target_trade_date, status, target_count, finished_at
FROM market_sync_batch
WHERE target_trade_date = (SELECT target_trade_date FROM latest_date)
ORDER BY
(status = 'running') DESC,
finished_at DESC NULLS LAST,
created_at DESC,
id DESC
LIMIT 1
"""
).fetchone()
if batch is None:
return MarketDataOverview(
status="no_data",
stock_pool_count=stock_pool_count,
latest_update=None,
)
batch_id = str(batch[0])
trade_date = batch[1]
batch_status = str(batch[2])
target_count = int(batch[3])
completed_at = batch[4]
stock_rows = cursor.execute(
"""
SELECT
stock.ts_code,
stock.name,
EXISTS (
SELECT 1
FROM market_daily_bar AS bar
WHERE bar.ts_code = stock.ts_code
AND bar.trade_date = %s
) AS has_bar,
EXISTS (
SELECT 1
FROM market_daily_basic AS basic
WHERE basic.ts_code = stock.ts_code
AND basic.trade_date = %s
) AS has_basic,
item.error_type,
item.error_message
FROM market_stock AS stock
LEFT JOIN market_sync_item AS item
ON item.batch_id = %s
AND item.item_key = stock.ts_code
AND item.item_kind = 'bar'
AND item.status = 'failed'
WHERE stock.is_active = true
ORDER BY stock.ts_code
""",
(trade_date, trade_date, batch_id),
).fetchall()
failures = tuple(
self._overview_failure(row, trade_date)
for row in stock_rows
if not (bool(row[2]) and bool(row[3]))
)
successful_count = len(stock_rows) - len(failures)
failed_count = max(0, stock_pool_count - successful_count)
batch_error_rows = cursor.execute(
"""
SELECT item_key, error_type, error_message
FROM market_sync_item
WHERE batch_id = %s
AND item_kind = 'batch'
AND status = 'failed'
ORDER BY item_key
""",
(batch_id,),
).fetchall()
batch_errors = tuple(
OverviewBatchError(
item_key=str(row[0]),
error_type=str(row[1] or "batch_error"),
reason=str(row[2] or "同步批次发生批次级错误"),
)
for row in batch_error_rows
)
coverage_rate = (
Decimal(successful_count) / Decimal(target_count) if target_count > 0 else Decimal("0")
)
latest_status = self._overview_status(
batch_status=batch_status,
successful_count=successful_count,
target_count=target_count,
failures=failures,
batch_errors=batch_errors,
)
return MarketDataOverview(
status=latest_status,
stock_pool_count=stock_pool_count,
latest_update=MarketDataUpdate(
trade_date=trade_date,
completed_at=completed_at,
status=latest_status,
target_count=target_count,
successful_count=successful_count,
failed_count=failed_count,
valid_count=successful_count,
coverage_rate=coverage_rate,
failures=failures,
batch_errors=batch_errors,
),
)
@staticmethod
def _overview_failure(row: tuple[Any, ...], trade_date: date) -> OverviewFailure:
"""Translate one incomplete stock row into a stable user-facing reason."""
ts_code = str(row[0])
name = str(row[1])
has_bar = bool(row[2])
has_basic = bool(row[3])
recorded_error_type = row[4]
recorded_message = row[5]
if recorded_error_type or recorded_message:
return OverviewFailure(
ts_code=ts_code,
name=name,
failure_type="recorded_stock_failure",
reason=str(recorded_message or "股票级同步失败"),
trade_date=trade_date,
)
if not has_bar and not has_basic:
return OverviewFailure(
ts_code=ts_code,
name=name,
failure_type="missing_daily_bar_and_valuation_snapshot",
reason="目标交易日缺少日线行情和估值快照",
trade_date=trade_date,
)
if not has_bar:
return OverviewFailure(
ts_code=ts_code,
name=name,
failure_type="missing_daily_bar",
reason="目标交易日缺少日线行情",
trade_date=trade_date,
)
return OverviewFailure(
ts_code=ts_code,
name=name,
failure_type="missing_valuation_snapshot",
reason="目标交易日缺少估值快照",
trade_date=trade_date,
)
@staticmethod
def _overview_status(
*,
batch_status: str,
successful_count: int,
target_count: int,
failures: tuple[OverviewFailure, ...],
batch_errors: tuple[OverviewBatchError, ...],
) -> LatestUpdateStatus:
"""Map persisted batch state and current fact completeness to a status."""
if batch_status == "running":
return "updating"
if batch_status == "failed" or (batch_errors and successful_count == 0):
return "failed"
if successful_count == 0:
return "failed"
if (
batch_status == "success"
and not failures
and not batch_errors
and successful_count >= target_count
):
return "success"
return "partial_success"
def upsert_stocks(self, rows: Iterable[Stock]) -> WriteResult:
records = tuple(rows)
if not records:
@@ -0,0 +1,153 @@
"""HTTP presentation for the Home market-data overview."""
from datetime import date, datetime
from typing import Annotated, Literal
from zoneinfo import ZoneInfo
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from zhixing_server.bootstrap.config import Settings, get_settings
from zhixing_server.modules.market_data.application.overview import (
MarketDataOverviewReader,
get_market_data_overview,
)
from zhixing_server.modules.market_data.domain.overview import (
MarketDataOverview,
MarketDataUpdate,
OverviewBatchError,
OverviewFailure,
)
from zhixing_server.modules.market_data.infrastructure.postgres import (
PostgresMarketDataRepository,
)
home_router = APIRouter()
_SHANGHAI = ZoneInfo("Asia/Shanghai")
OverviewStatusValue = Literal["updating", "success", "partial_success", "failed", "no_data"]
UpdateStatusValue = Literal["updating", "success", "partial_success", "failed"]
class OverviewFailureResponse(BaseModel):
"""A stock-level failure in the public Home response."""
ts_code: str
name: str
failure_type: str
reason: str
trade_date: date
@classmethod
def from_domain(cls, failure: OverviewFailure) -> "OverviewFailureResponse":
"""Convert a domain failure without changing its business date."""
return cls(
ts_code=failure.ts_code,
name=failure.name,
failure_type=failure.failure_type,
reason=failure.reason,
trade_date=failure.trade_date,
)
class OverviewBatchErrorResponse(BaseModel):
"""A batch-level error kept separate from stock failures."""
item_key: str
error_type: str
reason: str
@classmethod
def from_domain(cls, error: OverviewBatchError) -> "OverviewBatchErrorResponse":
"""Convert a batch error to the public response shape."""
return cls(
item_key=error.item_key,
error_type=error.error_type,
reason=error.reason,
)
class LatestUpdateResponse(BaseModel):
"""The selected synchronization result for one target trading date."""
trade_date: date
completed_at: datetime | None
status: UpdateStatusValue
target_count: int
successful_count: int
failed_count: int
valid_count: int
coverage_rate: float
failures: list[OverviewFailureResponse]
batch_errors: list[OverviewBatchErrorResponse]
@classmethod
def from_domain(cls, update: MarketDataUpdate) -> "LatestUpdateResponse":
"""Convert and localize the synchronization completion timestamp."""
return cls(
trade_date=update.trade_date,
completed_at=_to_shanghai(update.completed_at),
status=update.status,
target_count=update.target_count,
successful_count=update.successful_count,
failed_count=update.failed_count,
valid_count=update.valid_count,
coverage_rate=float(update.coverage_rate),
failures=[OverviewFailureResponse.from_domain(item) for item in update.failures],
batch_errors=[
OverviewBatchErrorResponse.from_domain(item) for item in update.batch_errors
],
)
class HomeOverviewResponse(BaseModel):
"""Stable HTTP contract consumed by the Home feature."""
status: OverviewStatusValue
stock_pool_count: int
latest_update: LatestUpdateResponse | None
@classmethod
def from_domain(cls, overview: MarketDataOverview) -> "HomeOverviewResponse":
"""Convert the domain overview into its Pydantic response model."""
return cls(
status=overview.status,
stock_pool_count=overview.stock_pool_count,
latest_update=(
LatestUpdateResponse.from_domain(overview.latest_update)
if overview.latest_update is not None
else None
),
)
def _to_shanghai(value: datetime | None) -> datetime | None:
"""Render database timestamps in the product's business timezone."""
if value is None:
return None
if value.tzinfo is None:
value = value.replace(tzinfo=_SHANGHAI)
return value.astimezone(_SHANGHAI)
def get_market_data_overview_reader(
settings: Annotated[Settings, Depends(get_settings)],
) -> MarketDataOverviewReader:
"""Build the PostgreSQL read adapter for one request."""
return PostgresMarketDataRepository(settings.database_url)
@home_router.get("/overview", response_model=HomeOverviewResponse)
def market_data_overview(
reader: Annotated[MarketDataOverviewReader, Depends(get_market_data_overview_reader)],
) -> HomeOverviewResponse:
"""Return the read-only market-data overview for the Home page."""
overview = get_market_data_overview(reader)
return HomeOverviewResponse.from_domain(overview)
@@ -0,0 +1,219 @@
"""HTTP contract tests for the Home market-data overview."""
from datetime import UTC, date, datetime
from decimal import Decimal
from fastapi.testclient import TestClient
from zhixing_server.bootstrap.app import create_app
from zhixing_server.modules.market_data.application.overview import (
MarketDataOverviewReader,
)
from zhixing_server.modules.market_data.domain.overview import (
MarketDataOverview,
MarketDataUpdate,
OverviewBatchError,
OverviewFailure,
)
from zhixing_server.modules.market_data.presentation.home import (
get_market_data_overview_reader,
)
TARGET_DATE = date(2026, 8, 6)
class FakeOverviewReader:
"""Small in-memory read adapter used at the HTTP dependency seam."""
def __init__(self, overview: MarketDataOverview) -> None:
self.overview = overview
def read_overview(self) -> MarketDataOverview:
return self.overview
def get_client(overview: MarketDataOverview) -> TestClient:
"""Build the real application with only its read adapter overridden."""
application = create_app()
reader = FakeOverviewReader(overview)
def override_reader() -> MarketDataOverviewReader:
return reader
application.dependency_overrides[get_market_data_overview_reader] = override_reader
return TestClient(application)
def test_home_overview_returns_complete_success_contract() -> None:
"""A complete batch exposes active-pool counts and Shanghai completion time."""
response = get_client(
MarketDataOverview(
status="success",
stock_pool_count=2,
latest_update=MarketDataUpdate(
trade_date=TARGET_DATE,
completed_at=datetime(2026, 8, 7, 8, 30, tzinfo=UTC),
status="success",
target_count=2,
successful_count=2,
failed_count=0,
valid_count=2,
coverage_rate=Decimal("1"),
failures=(),
batch_errors=(),
),
)
).get("/api/v1/home/overview")
assert response.status_code == 200
assert response.json() == {
"status": "success",
"stock_pool_count": 2,
"latest_update": {
"trade_date": "2026-08-06",
"completed_at": "2026-08-07T16:30:00+08:00",
"status": "success",
"target_count": 2,
"successful_count": 2,
"failed_count": 0,
"valid_count": 2,
"coverage_rate": 1.0,
"failures": [],
"batch_errors": [],
},
}
def test_home_overview_keeps_stock_and_batch_failures_separate() -> None:
"""Partial results preserve failure categories and batch-level errors."""
response = get_client(
MarketDataOverview(
status="partial_success",
stock_pool_count=3,
latest_update=MarketDataUpdate(
trade_date=TARGET_DATE,
completed_at=datetime(2026, 8, 6, 16, 30),
status="partial_success",
target_count=3,
successful_count=1,
failed_count=2,
valid_count=1,
coverage_rate=Decimal("0.3333333333"),
failures=(
OverviewFailure(
ts_code="000001.SZ",
name="平安银行",
failure_type="missing_daily_bar",
reason="目标交易日缺少日线行情",
trade_date=TARGET_DATE,
),
OverviewFailure(
ts_code="600000.SH",
name="浦发银行",
failure_type="recorded_stock_failure",
reason="供应商请求失败",
trade_date=TARGET_DATE,
),
),
batch_errors=(
OverviewBatchError(
item_key="retention",
error_type="database_error",
reason="保留策略执行失败",
),
),
),
)
).get("/api/v1/home/overview")
assert response.status_code == 200
payload = response.json()
assert payload["latest_update"]["failed_count"] == 2
assert payload["latest_update"]["coverage_rate"] == 0.3333333333
assert payload["latest_update"]["failures"][0]["failure_type"] == "missing_daily_bar"
assert payload["latest_update"]["batch_errors"] == [
{
"item_key": "retention",
"error_type": "database_error",
"reason": "保留策略执行失败",
}
]
def test_home_overview_returns_updating_without_completion_time() -> None:
"""A running batch remains distinguishable from an absent batch."""
response = get_client(
MarketDataOverview(
status="updating",
stock_pool_count=1,
latest_update=MarketDataUpdate(
trade_date=TARGET_DATE,
completed_at=None,
status="updating",
target_count=1,
successful_count=0,
failed_count=1,
valid_count=0,
coverage_rate=Decimal("0"),
failures=(),
batch_errors=(),
),
)
).get("/api/v1/home/overview")
assert response.status_code == 200
assert response.json()["status"] == "updating"
assert response.json()["latest_update"]["completed_at"] is None
def test_home_overview_returns_failed_batch_without_fabricating_stock_errors() -> None:
"""A failed batch keeps infrastructure errors separate from stock details."""
response = get_client(
MarketDataOverview(
status="failed",
stock_pool_count=2,
latest_update=MarketDataUpdate(
trade_date=TARGET_DATE,
completed_at=datetime(2026, 8, 6, 16, 30),
status="failed",
target_count=2,
successful_count=0,
failed_count=2,
valid_count=0,
coverage_rate=Decimal("0"),
failures=(),
batch_errors=(
OverviewBatchError(
item_key="retention",
error_type="database_error",
reason="保留策略执行失败",
),
),
),
)
).get("/api/v1/home/overview")
assert response.status_code == 200
assert response.json()["status"] == "failed"
assert response.json()["latest_update"]["failures"] == []
assert response.json()["latest_update"]["batch_errors"][0]["item_key"] == "retention"
def test_home_overview_returns_no_data_without_fake_update() -> None:
"""No synchronization batch is represented by a null latest update."""
response = get_client(
MarketDataOverview(status="no_data", stock_pool_count=7, latest_update=None)
).get("/api/v1/home/overview")
assert response.status_code == 200
assert response.json() == {
"status": "no_data",
"stock_pool_count": 7,
"latest_update": None,
}