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

fix(market-data): 增加同步进度日志并修复生产目录权限

Browse Source
This commit is contained in:
yuxuanhui
2026-08-07 00:13:49 +08:00
parent aa50b409c5
commit eabf102e2b
6 changed files with 236 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zhixing-server/Dockerfile
+2
View File
@@ -30,6 +30,8 @@ COPY migrations ./migrations
RUN uv sync --frozen --no-dev
RUN useradd --create-home --uid 10001 appuser
RUN mkdir -p /app/data/market-data \
&& chown -R 10001:10001 /app/data/market-data
USER appuser
ENV PATH="/app/.venv/bin:$PATH"
EXPOSE 8000
zhixing-server/src/zhixing_server/modules/market_data/application/sync.py
+152 -9
View File
@@ -2,6 +2,8 @@
from __future__ import annotations
import logging
import time
from collections.abc import Iterable, Sequence
from dataclasses import dataclass, field
from datetime import date, timedelta
@@ -15,6 +17,9 @@ from ..domain.rules import filter_current_hs_a_stocks
SyncMode = Literal["daily", "initialize", "retry"]
logger = logging.getLogger(__name__)
_PROGRESS_LOG_INTERVAL = 100
@dataclass(frozen=True, slots=True)
class SyncMarketDataCommand:
@@ -134,8 +139,20 @@ class SyncMarketData:
return self._execute_locked(command)
def _execute_locked(self, command: SyncMarketDataCommand) -> SyncBatchSummary:
started_at = time.monotonic()
logger.info(
"market_data_sync_started mode=%s requested_trade_date=%s",
command.mode,
command.target_trade_date or "auto",
)
target_trade_date = self._resolve_target(command.target_trade_date)
window = SyncWindow.from_target(target_trade_date)
logger.info(
"market_data_sync_target target_trade_date=%s window_start=%s window_end=%s",
target_trade_date,
window.start,
window.end,
)
all_stocks = filter_current_hs_a_stocks(self.source.fetch_stocks())
if not all_stocks:
return SyncBatchSummary(
@@ -158,6 +175,11 @@ class SyncMarketData:
command.parent_batch_id,
len(all_stocks),
)
logger.info(
"market_data_sync_batch_created batch_id=%s target_count=%d",
batch_id,
len(all_stocks),
)
failures: list[SyncFailure] = []
totals = [0, 0, 0]
stock_codes = {stock.ts_code for stock in all_stocks}
@@ -166,18 +188,71 @@ class SyncMarketData:
)
self._process_stock_master(batch_id, all_stocks, failures, totals)
self._log_progress(
batch_id=batch_id,
stage="stock_master",
current=1,
total=1,
item_key="current",
totals=totals,
failures=failures,
started_at=started_at,
force=bool(failures),
)
dates = self._dates_to_process(window, target_trade_date, command.mode, retry_items)
for trade_date in dates:
if (
command.mode == "retry"
and ("daily_basic", trade_date.isoformat()) not in retry_items
):
continue
logger.info(
"market_data_sync_stage_started batch_id=%s stage=daily_basic total=%d",
batch_id,
len(dates),
)
for current, trade_date in enumerate(dates, start=1):
failure_count = len(failures)
self._process_daily_basic(batch_id, trade_date, stock_codes, window, failures, totals)
for stock in all_stocks:
if command.mode == "retry" and ("bar", stock.ts_code) not in retry_items:
continue
self._log_progress(
batch_id=batch_id,
stage="daily_basic",
current=current,
total=len(dates),
item_key=trade_date.isoformat(),
totals=totals,
failures=failures,
started_at=started_at,
force=len(failures) > failure_count,
)
logger.info(
"market_data_sync_stage_completed batch_id=%s stage=daily_basic total=%d",
batch_id,
len(dates),
)
stocks_to_process = tuple(
stock
for stock in all_stocks
if command.mode != "retry" or ("bar", stock.ts_code) in retry_items
)
logger.info(
"market_data_sync_stage_started batch_id=%s stage=bar total=%d",
batch_id,
len(stocks_to_process),
)
for current, stock in enumerate(stocks_to_process, start=1):
failure_count = len(failures)
self._process_bar(batch_id, stock, window, failures, totals)
self._log_progress(
batch_id=batch_id,
stage="bar",
current=current,
total=len(stocks_to_process),
item_key=stock.ts_code,
totals=totals,
failures=failures,
started_at=started_at,
force=len(failures) > failure_count,
)
logger.info(
"market_data_sync_stage_completed batch_id=%s stage=bar total=%d",
batch_id,
len(stocks_to_process),
)
if not failures:
try:
@@ -205,6 +280,20 @@ class SyncMarketData:
status = "success" if not failures else "partial_success" if valid_count else "failed"
eligible = coverage >= self.coverage_threshold
self.repository.record_batch(batch_id, status, valid_count, coverage, eligible)
logger.info(
"market_data_sync_finished batch_id=%s status=%s target_count=%d valid_count=%d "
"coverage=%s failures=%d inserted=%d updated=%d unchanged=%d elapsed_seconds=%.1f",
batch_id,
status,
len(all_stocks),
valid_count,
coverage,
len(failures),
totals[0],
totals[1],
totals[2],
time.monotonic() - started_at,
)
return SyncBatchSummary(
batch_id,
target_trade_date,
@@ -283,6 +372,12 @@ class SyncMarketData:
self.snapshots.discard(staged)
failure = self._failure("stock", "current", exc)
failures.append(failure)
logger.warning(
"market_data_sync_item_failed batch_id=%s stage=stock_master item=current "
"error_type=%s",
batch_id,
failure.error_type,
)
self.repository.record_item(
batch_id,
"stock",
@@ -329,6 +424,13 @@ class SyncMarketData:
self.snapshots.discard(staged)
failure = self._failure("daily_basic", key, exc)
failures.append(failure)
logger.warning(
"market_data_sync_item_failed batch_id=%s stage=daily_basic item=%s "
"error_type=%s",
batch_id,
key,
failure.error_type,
)
self.repository.record_item(
batch_id,
"daily_basic",
@@ -381,6 +483,12 @@ class SyncMarketData:
self.snapshots.discard(staged)
failure = self._failure("bar", stock.ts_code, exc)
failures.append(failure)
logger.warning(
"market_data_sync_item_failed batch_id=%s stage=bar item=%s error_type=%s",
batch_id,
stock.ts_code,
failure.error_type,
)
self.repository.record_item(
batch_id,
"bar",
@@ -397,6 +505,41 @@ class SyncMarketData:
totals[1] += result.updated
totals[2] += result.unchanged
@staticmethod
def _log_progress(
*,
batch_id: str,
stage: str,
current: int,
total: int,
item_key: str,
totals: list[int],
failures: Sequence[SyncFailure],
started_at: float,
force: bool = False,
) -> None:
"""Log bounded, secret-free progress for a batch stage."""
if not force and total > _PROGRESS_LOG_INTERVAL and current not in {
1,
total,
} and current % _PROGRESS_LOG_INTERVAL != 0:
return
logger.info(
"market_data_sync_progress batch_id=%s stage=%s progress=%d/%d item=%s "
"failures=%d inserted=%d updated=%d unchanged=%d elapsed_seconds=%.1f",
batch_id,
stage,
current,
total,
item_key,
len(failures),
totals[0],
totals[1],
totals[2],
time.monotonic() - started_at,
)
@staticmethod
def _failure(item_kind: str, item_key: str, error: BaseException) -> SyncFailure:
message = " ".join(str(error).split())[:500] or "synchronization item failed"
zhixing-server/src/zhixing_server/modules/market_data/infrastructure/tushare.py
+14
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import logging
import random
import time
from collections.abc import Callable, Iterable, Mapping
@@ -11,6 +12,8 @@ from typing import cast
from ..domain.models import Bar, DailyBasic, Stock, SyncWindow, parse_date
from ..domain.rules import filter_current_hs_a_stocks
logger = logging.getLogger(__name__)
class TushareSourceError(RuntimeError):
"""A vendor request failed after the configured retry budget."""
@@ -158,7 +161,18 @@ class TushareAdapter:
if attempt == self.max_retries:
break
delay = self.backoff_seconds * (2**attempt) * (0.5 + self.random_fn())
logger.warning(
"tushare_request_retry method=%s attempt=%d max_attempts=%d",
method_name,
attempt + 1,
self.max_retries + 1,
)
self.sleep_fn(delay)
logger.error(
"tushare_request_failed method=%s attempts=%d",
method_name,
self.max_retries + 1,
)
raise TushareSourceError(f"Tushare request failed: {method_name}") from last_error
@staticmethod
zhixing-server/src/zhixing_server/modules/market_data/presentation/cli.py
+13
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import argparse
import json
import logging
from collections.abc import Sequence
from datetime import date
@@ -13,6 +14,8 @@ from ..infrastructure.csv_snapshot import CsvSnapshotStore
from ..infrastructure.postgres import PostgresMarketDataRepository
from ..infrastructure.tushare import TushareAdapter
logger = logging.getLogger(__name__)
def build_parser() -> argparse.ArgumentParser:
"""Build the explicit, repeatable synchronization CLI."""
@@ -40,6 +43,11 @@ def main(argv: Sequence[str] | None = None) -> int:
args = build_parser().parse_args(argv)
settings = get_settings()
logging.basicConfig(
level=settings.log_level.upper(),
format="%(asctime)s %(levelname)s %(name)s %(message)s",
force=True,
)
if args.retry_batch_id:
command = SyncMarketDataCommand(
mode="retry",
@@ -51,6 +59,11 @@ def main(argv: Sequence[str] | None = None) -> int:
mode="initialize" if args.initialize else "daily",
target_trade_date=args.trade_date,
)
logger.info(
"market_data_sync_cli mode=%s target_trade_date=%s",
command.mode,
command.target_trade_date or "auto",
)
source = TushareAdapter.from_token(
settings.tushare_token,
max_retries=settings.market_data_max_retries,
zhixing-server/tests/unit/market_data/test_sync.py
+38
View File
@@ -1,9 +1,12 @@
import logging
from collections.abc import Generator, Iterable, Sequence
from contextlib import contextmanager
from datetime import date
from decimal import Decimal
from pathlib import Path
import pytest
from zhixing_server.modules.market_data.application.sync import (
SyncMarketData,
SyncMarketDataCommand,
@@ -163,3 +166,38 @@ def test_sync_is_idempotent_and_reports_coverage(tmp_path: Path) -> None:
assert second.status == "success"
assert second.inserted_count == 1
assert second.unchanged_count >= 2
def test_sync_logs_progress_for_initialize_and_daily_update(
tmp_path: Path,
caplog: pytest.LogCaptureFixture,
) -> None:
target = date(2024, 1, 2)
repository = InMemoryRepository()
use_case = SyncMarketData(
FakeSource(target),
CsvSnapshotStore(tmp_path),
repository,
today=target,
)
caplog.set_level(logging.INFO, logger="zhixing_server.modules.market_data.application.sync")
initialize = use_case.execute(
SyncMarketDataCommand(mode="initialize", target_trade_date=target)
)
initialize_messages = [record.getMessage() for record in caplog.records]
assert initialize.status == "success"
assert any("stage=daily_basic" in message for message in initialize_messages)
assert any(
"stage=bar" in message and "progress=1/1" in message
for message in initialize_messages
)
caplog.clear()
daily = use_case.execute(SyncMarketDataCommand(mode="daily", target_trade_date=target))
daily_messages = [record.getMessage() for record in caplog.records]
assert daily.status == "success"
assert any("market_data_sync_started mode=daily" in message for message in daily_messages)
assert any("market_data_sync_finished" in message for message in daily_messages)