diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 40eb0ce..147fd39 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -51,11 +51,28 @@ services: context: ./zhixing-server target: production command: ["alembic", "upgrade", "head"] + depends_on: + market-data-init: + condition: service_completed_successfully environment: ZHIXING_DATABASE_URL: ${ZHIXING_DATABASE_URL:?Set ZHIXING_DATABASE_URL to the 1Panel PostgreSQL URL} networks: - 1panel-network + # Named volumes are initialized as root; normalize ownership before app jobs run. + market-data-init: + profiles: ["jobs"] + build: + context: ./zhixing-server + target: production + user: "0:0" + command: + - sh + - -c + - mkdir -p /app/data/market-data && chown -R 10001:10001 /app/data/market-data + volumes: + - market-data:/app/data/market-data + market-sync: profiles: ["jobs"] build: diff --git a/zhixing-server/Dockerfile b/zhixing-server/Dockerfile index 8593aae..bd755ac 100644 --- a/zhixing-server/Dockerfile +++ b/zhixing-server/Dockerfile @@ -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 diff --git a/zhixing-server/src/zhixing_server/modules/market_data/application/sync.py b/zhixing-server/src/zhixing_server/modules/market_data/application/sync.py index c143a18..b4a28c8 100644 --- a/zhixing-server/src/zhixing_server/modules/market_data/application/sync.py +++ b/zhixing-server/src/zhixing_server/modules/market_data/application/sync.py @@ -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" diff --git a/zhixing-server/src/zhixing_server/modules/market_data/infrastructure/tushare.py b/zhixing-server/src/zhixing_server/modules/market_data/infrastructure/tushare.py index f8f7be5..6a055f3 100644 --- a/zhixing-server/src/zhixing_server/modules/market_data/infrastructure/tushare.py +++ b/zhixing-server/src/zhixing_server/modules/market_data/infrastructure/tushare.py @@ -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 diff --git a/zhixing-server/src/zhixing_server/modules/market_data/presentation/cli.py b/zhixing-server/src/zhixing_server/modules/market_data/presentation/cli.py index cf0ec84..4e87ce5 100644 --- a/zhixing-server/src/zhixing_server/modules/market_data/presentation/cli.py +++ b/zhixing-server/src/zhixing_server/modules/market_data/presentation/cli.py @@ -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, diff --git a/zhixing-server/tests/unit/market_data/test_sync.py b/zhixing-server/tests/unit/market_data/test_sync.py index 83d13e8..44358fe 100644 --- a/zhixing-server/tests/unit/market_data/test_sync.py +++ b/zhixing-server/tests/unit/market_data/test_sync.py @@ -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)