From d8f0dd930c3eb61406464d826ed34bed038b6965 Mon Sep 17 00:00:00 2001 From: leefer Date: Thu, 30 Jul 2026 10:14:29 +0800 Subject: [PATCH] rebuild(runtime): govern market operations and job truth --- next/backend/bootstrap/application.py | 4 +- next/backend/bootstrap/container.py | 11 + next/backend/data/gateway.py | 23 ++ next/backend/data/providers/base.py | 9 + next/backend/data/providers/eastmoney.py | 11 + next/backend/data/providers/ifind.py | 116 +++++++ next/backend/data/providers/tushare.py | 62 ++++ next/backend/data/repository.py | 90 +++++ .../database/migrations/m0011_operations.py | 75 +++++ next/backend/database/migrations/registry.py | 2 + next/backend/features/market/events.py | 199 +++++++++++ next/backend/features/market/routes.py | 2 +- next/backend/features/market/service.py | 6 + next/backend/features/market/snapshot.py | 69 +++- next/backend/features/market/sync.py | 85 ++++- next/backend/features/operations/__init__.py | 3 + next/backend/features/operations/routes.py | 96 ++++++ next/backend/features/operations/schemas.py | 49 +++ next/backend/features/operations/service.py | 312 ++++++++++++++++++ next/backend/http/router.py | 2 + next/backend/jobs/operations.py | 25 ++ next/backend/jobs/repository.py | 154 +++++++++ next/backend/jobs/screener.py | 33 -- next/backend/jobs/service.py | 122 +++++++ next/docs/evidence/stage-15/acceptance.md | 16 +- next/docs/final/completion-audit.md | 24 +- next/frontend/src/app/shell/StatusBar.vue | 13 +- .../src/app/system/MarketSettingsPanel.vue | 123 ++++++- .../src/pages/screener/CandidateTable.vue | 16 +- .../src/pages/screener/StagePanel.vue | 20 +- .../src/pages/screener/StrategyPanel.vue | 14 +- next/frontend/src/shared/api/operations.ts | 60 ++++ next/frontend/src/shared/styles/mobile.css | 9 +- next/frontend/src/shared/styles/system.css | 24 ++ next/tests/e2e/operations.spec.js | 104 ++++++ next/tests/e2e/stage9.spec.js | 47 +++ next/tests/test_database_tool.py | 2 +- next/tests/test_migrations.py | 5 +- next/tests/test_operations.py | 266 +++++++++++++++ 39 files changed, 2224 insertions(+), 79 deletions(-) create mode 100644 next/backend/database/migrations/m0011_operations.py create mode 100644 next/backend/features/market/events.py create mode 100644 next/backend/features/operations/__init__.py create mode 100644 next/backend/features/operations/routes.py create mode 100644 next/backend/features/operations/schemas.py create mode 100644 next/backend/features/operations/service.py create mode 100644 next/backend/jobs/operations.py create mode 100644 next/backend/jobs/repository.py delete mode 100644 next/backend/jobs/screener.py create mode 100644 next/backend/jobs/service.py create mode 100644 next/frontend/src/shared/api/operations.ts create mode 100644 next/tests/e2e/operations.spec.js create mode 100644 next/tests/test_operations.py diff --git a/next/backend/bootstrap/application.py b/next/backend/bootstrap/application.py index 2f53877..52dbe95 100644 --- a/next/backend/bootstrap/application.py +++ b/next/backend/bootstrap/application.py @@ -13,7 +13,7 @@ from backend.http.errors import install_error_handlers from backend.http.request_context import install_request_context from backend.http.router import api_router from backend.http.security import install_security_headers -from backend.jobs.screener import run_screener_scheduler +from backend.jobs.operations import run_operations_scheduler logger = logging.getLogger(__name__) @@ -41,7 +41,7 @@ def create_application(settings: Settings | None = None) -> FastAPI: stop = asyncio.Event() task = None if runtime.environment != "test": - task = asyncio.create_task(run_screener_scheduler(container.screener, stop)) + task = asyncio.create_task(run_operations_scheduler(container.operations, stop)) try: yield finally: diff --git a/next/backend/bootstrap/container.py b/next/backend/bootstrap/container.py index 9e769e2..0b2be72 100644 --- a/next/backend/bootstrap/container.py +++ b/next/backend/bootstrap/container.py @@ -19,16 +19,20 @@ from backend.features.accounts.service import ( from backend.features.heaven.repository import HeavenRepository from backend.features.heaven.service import HeavenService from backend.features.market import MarketService +from backend.features.market.events import MarketEventService from backend.features.market.insights import MarketInsightService from backend.features.market.sync import MarketSnapshotService from backend.features.mentor.context import MentorContextBuilder from backend.features.mentor.repository import MentorRepository from backend.features.mentor.service import MentorService from backend.features.mentor.skills import MentorSkillRegistry +from backend.features.operations import OperationsService from backend.features.review import ReviewService from backend.features.review.repository import ReviewRepository from backend.features.screener.repository import ScreenerRepository from backend.features.screener.service import ScreenerService +from backend.jobs.repository import JobRepository +from backend.jobs.service import JobService from backend.llm import LLMGateway from backend.llm.repository import LLMRepository from backend.security import PasswordHasher, load_or_create_cipher @@ -49,6 +53,7 @@ class ApplicationContainer: mentor: MentorService heaven: HeavenService review: ReviewService + operations: OperationsService def build_container(settings: Settings) -> ApplicationContainer: @@ -104,6 +109,11 @@ def build_container(settings: Settings) -> ApplicationContainer: PROJECT_ROOT / "config" / "heaven" / "iching_zh.json", ) review = ReviewService(database, ReviewRepository(), gateway, screener, llm) + jobs = JobService(database, JobRepository()) + events = MarketEventService(database, market_repository, gateway) + operations = OperationsService( + database, market_repository, market, events, screener, jobs + ) return ApplicationContainer( settings=settings, database=database, @@ -118,4 +128,5 @@ def build_container(settings: Settings) -> ApplicationContainer: mentor=mentor, heaven=heaven, review=review, + operations=operations, ) diff --git a/next/backend/data/gateway.py b/next/backend/data/gateway.py index 354e24b..632505f 100644 --- a/next/backend/data/gateway.py +++ b/next/backend/data/gateway.py @@ -115,6 +115,19 @@ class DataGateway: self._policy.assert_allowed(provider.source, DataUsage.CALCULATION) return provider.snapshot_inputs(trade_date, previous_trade_date) + def realtime_snapshot_inputs( + self, trade_date: str, previous_trade_date: str + ) -> dict[str, Any]: + provider = self._provider(DataSource.TUSHARE) + self._policy.assert_allowed(provider.source, DataUsage.CALCULATION) + directory = self.stock_directory() + identifiers = tuple(sorted(directory)) + if not identifiers: + raise MarketDataUnavailable("股票目录为空，无法读取盘中行情") + return provider.realtime_market_inputs( + trade_date, previous_trade_date, identifiers + ) + def trading_dates(self, through: str, limit: int = 2) -> tuple[str, ...]: requested = _date(through) with self._database.read() as connection: @@ -163,6 +176,16 @@ class DataGateway: raise MarketDataUnavailable("当前动态竞价快照暂不可用") return result + def event_reasons(self, trade_date: str) -> ProviderResult: + provider = self._provider(DataSource.IFIND) + self._policy.assert_allowed(provider.source, DataUsage.CALCULATION) + result = provider.event_reasons(trade_date) + if result.metadata.usage != DataUsage.CALCULATION: + raise MarketDataUnavailable("事件补充来源未获准参与正式数据治理") + if result.metadata.coverage <= 0: + raise MarketDataUnavailable("事件原因补充服务暂不可用") + return result + def sector_members( self, trade_date: str, sector_name: str, representative: str ) -> dict[str, Any]: diff --git a/next/backend/data/providers/base.py b/next/backend/data/providers/base.py index ca90b30..ab4db82 100644 --- a/next/backend/data/providers/base.py +++ b/next/backend/data/providers/base.py @@ -27,6 +27,13 @@ class MarketDataProvider(Protocol): self, trade_date: str, previous_trade_date: str ) -> dict[str, ProviderResult | dict[str, Any]]: ... + def realtime_market_inputs( + self, + trade_date: str, + previous_trade_date: str, + identifiers: tuple[str, ...], + ) -> dict[str, ProviderResult | dict[str, Any]]: ... + def sector_members(self, representative: str, trade_date: str) -> ProviderResult: ... def heaven_inputs( @@ -49,4 +56,6 @@ class MarketDataProvider(Protocol): self, identifiers: tuple[str, ...], start_time: str, end_time: str ) -> ProviderResult: ... + def event_reasons(self, trade_date: str) -> ProviderResult: ... + def screener_inputs(self, trade_dates: tuple[str, ...]) -> dict[str, ProviderResult | None]: ... diff --git a/next/backend/data/providers/eastmoney.py b/next/backend/data/providers/eastmoney.py index c259c22..7d639cd 100644 --- a/next/backend/data/providers/eastmoney.py +++ b/next/backend/data/providers/eastmoney.py @@ -99,6 +99,14 @@ class EastmoneyProvider: ) -> dict[str, ProviderResult | dict[str, object]]: raise ProviderError("The display provider cannot build market snapshots") + def realtime_market_inputs( + self, + trade_date: str, + previous_trade_date: str, + identifiers: tuple[str, ...], + ) -> dict[str, ProviderResult | dict[str, object]]: + raise ProviderError("The display provider cannot build realtime market snapshots") + def sector_members(self, representative: str, trade_date: str) -> ProviderResult: raise ProviderError("The display provider is not the constituent authority") @@ -126,6 +134,9 @@ class EastmoneyProvider: ) -> ProviderResult: raise ProviderError("The display provider cannot supply calculation snapshots") + def event_reasons(self, trade_date: str) -> ProviderResult: + raise ProviderError("The display provider cannot supply event reasons") + def screener_inputs(self, trade_dates: tuple[str, ...]) -> dict[str, ProviderResult | None]: raise ProviderError("The display provider cannot supply screener factors") diff --git a/next/backend/data/providers/ifind.py b/next/backend/data/providers/ifind.py index 0525e42..7160bb5 100644 --- a/next/backend/data/providers/ifind.py +++ b/next/backend/data/providers/ifind.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import re import threading import urllib.error import urllib.request @@ -89,6 +90,14 @@ class IfindProvider: ) -> dict[str, ProviderResult | dict[str, Any]]: raise ProviderError("iFinD is not the post-close snapshot authority") + def realtime_market_inputs( + self, + trade_date: str, + previous_trade_date: str, + identifiers: tuple[str, ...], + ) -> dict[str, ProviderResult | dict[str, Any]]: + raise ProviderError("iFinD不承担全市场盘中快照计算") + def sector_members(self, representative: str, trade_date: str) -> ProviderResult: raise ProviderError("iFinD is not the Shenwan constituent authority") @@ -145,6 +154,76 @@ class IfindProvider: ), ) + def event_reasons(self, trade_date: str) -> ProviderResult: + date = datetime.fromisoformat(_display(trade_date)).date() + display = f"{date.year}年{date.month}月{date.day}日" + queries = { + "limit_up": ( + f"{display}涨停股票，股票代码、股票简称、涨停原因、" + "首次涨停时间、最终涨停时间、开板次数" + ), + "broken": ( + f"{display}曾涨停但收盘未涨停的股票，股票代码、股票简称、" + "涨停原因、首次涨停时间、开板次数" + ), + "limit_down": f"{display}跌停股票，股票代码、股票简称、跌停原因", + } + rows: list[dict[str, Any]] = [] + completed = 0 + for event_type, query in queries.items(): + try: + payload = self._request( + "smart_stock_picking", + {"searchstring": query, "searchtype": "stock"}, + ) + except ProviderError: + continue + completed += 1 + for raw in _result( + payload, "event", "not_applicable", SnapshotState.FINAL + ).rows: + identifier = _event_identifier(raw) + if not identifier: + continue + reason_tokens = ( + ("跌停原因", "风险线索", "原因") + if event_type == "limit_down" + else ("涨停原因类别", "涨停原因", "触板逻辑", "原因") + ) + rows.append( + { + "event_type": event_type, + "identifier": identifier, + "reason": str(_event_field(raw, reason_tokens) or "").strip(), + "first_time": _event_time( + _event_field( + raw, ("首次涨停时间", "首次触板时间", "首次封板时间") + ) + ), + "last_time": _event_time( + _event_field( + raw, ("最终涨停时间", "最后涨停时间", "最后封板时间") + ) + ), + "open_times": _event_integer( + _event_field(raw, ("开板次数", "打开涨停次数")) + ), + } + ) + return ProviderResult( + tuple(rows), + ObservationMetadata( + source=self.source, + observed_at=datetime.now(SHANGHAI), + unit="event", + adjustment="not_applicable", + freshness_seconds=0, + coverage=completed / len(queries), + state=SnapshotState.FINAL, + usage=DataUsage.CALCULATION, + ), + ) + def screener_inputs(self, trade_dates: tuple[str, ...]) -> dict[str, ProviderResult | None]: raise ProviderError("iFinD尚未批准用于盘后因子批量计算") @@ -295,3 +374,40 @@ def _error_code(payload: dict[str, Any]) -> int: return int(payload.get("errorcode", payload.get("code", 0)) or 0) except (TypeError, ValueError): return -1 + + +def _event_field(row: dict[str, Any], tokens: tuple[str, ...]) -> Any: + for token in tokens: + for key, value in row.items(): + if token in str(key) and value not in (None, ""): + return value + return None + + +def _event_identifier(row: dict[str, Any]) -> str: + value = _event_field(row, ("股票代码", "证券代码", "代码", "thscode")) + text = str(value or "").strip().upper() + match = re.search(r"(? str: + text = str(value or "").strip() + match = re.search(r"(? int | None: + try: + return max(0, int(float(value))) if value not in (None, "") else None + except (TypeError, ValueError): + return None diff --git a/next/backend/data/providers/tushare.py b/next/backend/data/providers/tushare.py index 9d77dc2..7bd9ae9 100644 --- a/next/backend/data/providers/tushare.py +++ b/next/backend/data/providers/tushare.py @@ -130,6 +130,59 @@ class TushareProvider: ) return datasets + def realtime_market_inputs( + self, + trade_date: str, + previous_trade_date: str, + identifiers: tuple[str, ...], + ) -> dict[str, ProviderResult | dict[str, Any]]: + current = _compact(trade_date) + previous = _compact(previous_trade_date) + quotes = self._query( + "rt_k", + {"ts_code": ",".join(identifiers)}, + "", + unit="mixed", + ) + normalized = tuple( + { + **row, + "trade_date": current, + "pct_chg": _change(row.get("close"), row.get("pre_close")), + "amount_unit": "yuan", + } + for row in quotes.rows + if _number(row.get("close")) > 0 and _number(row.get("pre_close")) > 0 + ) + daily = ProviderResult( + normalized, + replace( + quotes.metadata, + coverage=min(len(normalized) / max(len(identifiers), 1), 1), + state=SnapshotState.REALTIME, + ), + ) + return { + "daily": daily, + "price_limits": self._query( + "stk_limit", + {"trade_date": current}, + "ts_code,trade_date,up_limit,down_limit", + unit="yuan/share", + ), + "previous_limit_up": self._query( + "limit_list_d", + {"trade_date": previous, "limit_type": "U"}, + ( + "trade_date,ts_code,industry,name,close,pct_chg,amount,limit_amount," + "float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time," + "open_times,up_stat,limit_times" + ), + unit="mixed", + empty_is_complete=True, + ), + } + def sector_members(self, representative: str, trade_date: str) -> ProviderResult: target = _compact(trade_date) industry, members = self._sector_memberships(representative, target) @@ -354,6 +407,9 @@ class TushareProvider: ) -> ProviderResult: raise ProviderError("Tushare不提供动态竞价快照") + def event_reasons(self, trade_date: str) -> ProviderResult: + raise ProviderError("Tushare涨跌停榜单不提供可用的事件原因字段") + def screener_inputs(self, trade_dates: tuple[str, ...]) -> dict[str, ProviderResult | None]: if len(trade_dates) < 21: raise ProviderError("选股因子至少需要21个交易日") @@ -628,6 +684,12 @@ def _display(value: str) -> str: return f"{compact[:4]}-{compact[4:6]}-{compact[6:]}" +def _change(close: Any, previous: Any) -> float: + current = _number(close) + prior = _number(previous) + return round((current / prior - 1) * 100, 4) if current > 0 and prior > 0 else 0.0 + + def _quarter_periods(through: str, count: int) -> tuple[str, ...]: year = int(through[:4]) quarter = (int(through[4:6]) - 1) // 3 diff --git a/next/backend/data/repository.py b/next/backend/data/repository.py index 8f513b1..1fc83a7 100644 --- a/next/backend/data/repository.py +++ b/next/backend/data/repository.py @@ -163,6 +163,21 @@ class MarketRepository: ) ) + def open_dates_between( + self, connection: sqlite3.Connection, start_date: str, end_date: str + ) -> tuple[str, ...]: + return tuple( + str(row["trade_date"]) + for row in connection.execute( + """ + SELECT trade_date FROM trading_days + WHERE is_open = 1 AND trade_date BETWEEN ? AND ? + ORDER BY trade_date + """, + (start_date, end_date), + ).fetchall() + ) + def active_stock_count(self, connection: sqlite3.Connection) -> int: row = connection.execute( """ @@ -235,6 +250,81 @@ class MarketRepository: (through,), ).fetchone() + def save_event_revision( + self, + connection: sqlite3.Connection, + *, + trade_date: str, + identifier: str, + event_type: str, + reason: str, + first_time: str, + last_time: str, + open_times: int | None, + source: str, + priority: int, + created_by: int | None, + created_at: str, + ) -> int: + cursor = connection.execute( + """ + INSERT INTO market_event_revisions ( + trade_date, identifier, event_type, reason, first_time, last_time, + open_times, source, priority, created_by, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + trade_date, + identifier, + event_type, + reason, + first_time, + last_time, + open_times, + source, + priority, + created_by, + created_at, + ), + ) + return int(cursor.lastrowid) + + def event_revisions( + self, connection: sqlite3.Connection, trade_date: str + ) -> tuple[sqlite3.Row, ...]: + return tuple( + connection.execute( + """ + SELECT * FROM ( + SELECT revisions.*, + ROW_NUMBER() OVER ( + PARTITION BY identifier, event_type + ORDER BY priority DESC, id DESC + ) AS rank + FROM market_event_revisions AS revisions + WHERE trade_date = ? + ) WHERE rank = 1 + """, + (trade_date,), + ).fetchall() + ) + + def event_revision_history( + self, connection: sqlite3.Connection, trade_date: str, identifier: str + ) -> tuple[sqlite3.Row, ...]: + return tuple( + connection.execute( + """ + SELECT revisions.*, users.username AS created_by_name + FROM market_event_revisions AS revisions + LEFT JOIN users ON users.id = revisions.created_by + WHERE revisions.trade_date = ? AND revisions.identifier = ? + ORDER BY revisions.id DESC + """, + (trade_date, identifier), + ).fetchall() + ) + def sector_members( self, connection: sqlite3.Connection, trade_date: str, sector_name: str ) -> sqlite3.Row | None: diff --git a/next/backend/database/migrations/m0011_operations.py b/next/backend/database/migrations/m0011_operations.py new file mode 100644 index 0000000..1e9f839 --- /dev/null +++ b/next/backend/database/migrations/m0011_operations.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import sqlite3 + +from backend.database.migrations.runner import Migration + + +def upgrade(connection: sqlite3.Connection) -> None: + statements = ( + """ + CREATE TABLE job_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kind TEXT NOT NULL, + run_key TEXT NOT NULL, + requested_date TEXT NOT NULL DEFAULT '', + trigger TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('running','completed','failed')), + attempt INTEGER NOT NULL CHECK (attempt >= 1), + started_at TEXT NOT NULL, + finished_at TEXT, + duration_ms INTEGER, + coverage REAL CHECK (coverage BETWEEN 0 AND 1), + source_set_json TEXT NOT NULL DEFAULT '[]', + output_version TEXT NOT NULL DEFAULT '', + payload_json TEXT NOT NULL DEFAULT '{}', + error_code TEXT NOT NULL DEFAULT '', + error_message TEXT NOT NULL DEFAULT '', + UNIQUE(kind, run_key, attempt) + ) + """, + """ + CREATE UNIQUE INDEX job_runs_active_kind_idx + ON job_runs(kind) WHERE status = 'running' + """, + """ + CREATE INDEX job_runs_kind_started_idx + ON job_runs(kind, started_at DESC, id DESC) + """, + """ + CREATE TABLE market_event_revisions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + trade_date TEXT NOT NULL, + identifier TEXT NOT NULL, + event_type TEXT NOT NULL CHECK (event_type IN ('limit_up','broken','limit_down')), + reason TEXT NOT NULL DEFAULT '', + first_time TEXT NOT NULL DEFAULT '', + last_time TEXT NOT NULL DEFAULT '', + open_times INTEGER CHECK (open_times IS NULL OR open_times >= 0), + source TEXT NOT NULL CHECK (source IN ('tushare','ifind','admin')), + priority INTEGER NOT NULL CHECK (priority >= 0), + created_by INTEGER REFERENCES users(id) ON DELETE SET NULL, + created_at TEXT NOT NULL + ) + """, + """ + CREATE INDEX market_event_revisions_lookup_idx + ON market_event_revisions(trade_date, identifier, event_type, priority DESC, id DESC) + """, + ) + for statement in statements: + connection.execute(statement) + + +def downgrade(connection: sqlite3.Connection) -> None: + connection.execute("DROP TABLE market_event_revisions") + connection.execute("DROP TABLE job_runs") + + +MIGRATION = Migration( + version=11, + name="create_operations_audit", + signature="operations:v1:jobs-event-revisions", + upgrade=upgrade, + downgrade=downgrade, +) diff --git a/next/backend/database/migrations/registry.py b/next/backend/database/migrations/registry.py index 23d0c15..ac0b4c0 100644 --- a/next/backend/database/migrations/registry.py +++ b/next/backend/database/migrations/registry.py @@ -8,6 +8,7 @@ from backend.database.migrations.m0007_screener import MIGRATION as SCREENER from backend.database.migrations.m0008_mentor_llm import MIGRATION as MENTOR_LLM from backend.database.migrations.m0009_heaven import MIGRATION as HEAVEN from backend.database.migrations.m0010_review import MIGRATION as REVIEW +from backend.database.migrations.m0011_operations import MIGRATION as OPERATIONS from backend.database.migrations.runner import Migration MIGRATIONS: tuple[Migration, ...] = ( @@ -21,4 +22,5 @@ MIGRATIONS: tuple[Migration, ...] = ( MENTOR_LLM, HEAVEN, REVIEW, + OPERATIONS, ) diff --git a/next/backend/features/market/events.py b/next/backend/features/market/events.py new file mode 100644 index 0000000..16a45c8 --- /dev/null +++ b/next/backend/features/market/events.py @@ -0,0 +1,199 @@ +from __future__ import annotations + +import json +from datetime import date, datetime +from typing import Any +from zoneinfo import ZoneInfo + +from backend.data.gateway import DataGateway +from backend.data.repository import MarketRepository +from backend.database.connection import Database + +SHANGHAI = ZoneInfo("Asia/Shanghai") +EVENT_KEYS = {"limit_up": "limits", "broken": "broken", "limit_down": "down_limits"} + + +class MarketEventError(RuntimeError): + pass + + +class MarketEventService: + def __init__( + self, database: Database, repository: MarketRepository, gateway: DataGateway + ) -> None: + self._database = database + self._repository = repository + self._gateway = gateway + + def supplement(self, trade_date: str) -> dict[str, Any]: + target = _date(trade_date) + snapshot = self._snapshot(target) + expected = { + (str(row.get("identifier") or ""), event_type) + for event_type, key in EVENT_KEYS.items() + for row in snapshot.get(key) or [] + if row.get("identifier") + } + if not expected: + return { + "trade_date": target, + "events": 0, + "updated": 0, + "coverage": 1.0, + "source_set": ["local"], + "output_version": "event-revisions-v1", + } + result = self._gateway.event_reasons(target) + now = datetime.now(SHANGHAI).isoformat(timespec="seconds") + updated = 0 + matched: set[tuple[str, str]] = set() + with self._database.transaction() as connection: + current = { + (str(row["identifier"]), str(row["event_type"])): dict(row) + for row in self._repository.event_revisions(connection, target) + } + for row in result.rows: + identity = (str(row.get("identifier") or ""), str(row.get("event_type") or "")) + if identity not in expected: + continue + useful = any( + row.get(field) not in (None, "") + for field in ("reason", "first_time", "last_time", "open_times") + ) + if not useful: + continue + matched.add(identity) + values = _revision_values(row) + existing = current.get(identity) + if existing and int(existing["priority"]) >= 100: + continue + if existing and all(existing.get(key) == value for key, value in values.items()): + continue + self._repository.save_event_revision( + connection, + trade_date=target, + identifier=identity[0], + event_type=identity[1], + source="ifind", + priority=20, + created_by=None, + created_at=now, + **values, + ) + updated += 1 + return { + "trade_date": target, + "events": len(expected), + "matched": len(matched), + "updated": updated, + "coverage": round(len(matched) / len(expected), 4), + "source_set": ["ifind", "local"], + "output_version": "event-revisions-v1", + } + + def revise( + self, + *, + trade_date: str, + identifier: str, + event_type: str, + reason: str, + first_time: str, + last_time: str, + open_times: int | None, + user_id: int, + ) -> dict[str, Any]: + target = _date(trade_date) + normalized = identifier.strip().upper() + if event_type not in EVENT_KEYS: + raise MarketEventError("事件类型无效") + snapshot = self._snapshot(target) + exists = any( + str(row.get("identifier") or "") == normalized + for row in snapshot.get(EVENT_KEYS[event_type]) or [] + ) + if not exists: + raise MarketEventError("该股票不在所选日期的对应事件池中") + normalized_reason = " ".join(reason.split()) + if not 1 <= len(normalized_reason) <= 200: + raise MarketEventError("原因应为1至200个字符") + values = { + "reason": normalized_reason, + "first_time": _time(first_time), + "last_time": _time(last_time), + "open_times": open_times, + } + with self._database.transaction() as connection: + revision_id = self._repository.save_event_revision( + connection, + trade_date=target, + identifier=normalized, + event_type=event_type, + source="admin", + priority=100, + created_by=user_id, + created_at=datetime.now(SHANGHAI).isoformat(timespec="seconds"), + **values, + ) + return {"id": revision_id, "trade_date": target, "identifier": normalized, **values} + + def history(self, trade_date: str, identifier: str) -> list[dict[str, Any]]: + target = _date(trade_date) + with self._database.read() as connection: + rows = self._repository.event_revision_history( + connection, target, identifier.strip().upper() + ) + return [dict(row) for row in rows] + + def _snapshot(self, trade_date: str) -> dict[str, Any]: + with self._database.read() as connection: + row = self._repository.latest_summary(connection, trade_date) + if row is None or str(row["trade_date"]) != trade_date: + raise MarketEventError("所选日期没有正式行情快照") + return json.loads(str(row["payload_json"])) + + +def apply_event_revisions( + payload: dict[str, Any], revisions: tuple[Any, ...] +) -> dict[str, Any]: + index = { + (str(row["identifier"]), str(row["event_type"])): row for row in revisions + } + for event_type, key in EVENT_KEYS.items(): + for item in payload.get(key) or []: + revision = index.get((str(item.get("identifier") or ""), event_type)) + if revision is None: + continue + for field in ("reason", "first_time", "last_time", "open_times"): + value = revision[field] + if value not in (None, ""): + item[field] = value + item["reason_source"] = str(revision["source"]) + item["revision_id"] = int(revision["id"]) + return payload + + +def _revision_values(row: dict[str, Any]) -> dict[str, Any]: + return { + "reason": str(row.get("reason") or "").strip(), + "first_time": _time(str(row.get("first_time") or "")), + "last_time": _time(str(row.get("last_time") or "")), + "open_times": row.get("open_times") if isinstance(row.get("open_times"), int) else None, + } + + +def _date(value: str) -> str: + try: + return date.fromisoformat(value).isoformat() + except ValueError as exc: + raise MarketEventError("日期格式无效") from exc + + +def _time(value: str) -> str: + normalized = value.strip() + if not normalized: + return "" + try: + return datetime.strptime(normalized, "%H:%M").strftime("%H:%M") + except ValueError as exc: + raise MarketEventError("事件时间格式应为HH:MM") from exc diff --git a/next/backend/features/market/routes.py b/next/backend/features/market/routes.py index 189922a..a02c261 100644 --- a/next/backend/features/market/routes.py +++ b/next/backend/features/market/routes.py @@ -86,7 +86,7 @@ def sync_snapshot( _principal: AdminWritePrincipal, requested_date: Annotated[str | None, Query(alias="date")] = None, ) -> dict: - return request.app.state.container.market.sync_snapshot(requested_date) + return request.app.state.container.operations.refresh(requested_date) @router.get("/workspaces/{key}", response_model=MarketWorkspaceResponse) diff --git a/next/backend/features/market/service.py b/next/backend/features/market/service.py index 2bb79b2..a5ace35 100644 --- a/next/backend/features/market/service.py +++ b/next/backend/features/market/service.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import datetime from typing import Any from backend.data.gateway import DataGateway, MarketDataUnavailable @@ -96,6 +97,11 @@ class MarketService: def sync_snapshot(self, requested_date: str | None = None) -> dict[str, Any]: return self._call(self._snapshots.sync, requested_date) + def sync_realtime( + self, requested_date: str | None = None, now: datetime | None = None + ) -> dict[str, Any]: + return self._call(self._snapshots.sync_realtime, requested_date, now) + def workspace(self, key: str, requested_date: str | None = None) -> dict[str, Any]: return self._call(self._snapshots.workspace, key, requested_date) diff --git a/next/backend/features/market/snapshot.py b/next/backend/features/market/snapshot.py index 22eaede..2997952 100644 --- a/next/backend/features/market/snapshot.py +++ b/next/backend/features/market/snapshot.py @@ -34,7 +34,7 @@ def build_snapshot( up_count = sum(_number(row.get("pct_chg")) > 0 for row in daily_rows) down_count = sum(_number(row.get("pct_chg")) < 0 for row in daily_rows) flat_count = len(daily_rows) - up_count - down_count - amount = sum(_number(row.get("amount")) * 1000 for row in daily_rows) + amount = sum(_amount_yuan(row) for row in daily_rows) seal_rate = len(limits) / max(len(limits) + len(broken), 1) * 100 sectors = _sectors(limits) previous_sectors = _sectors(previous_limits) @@ -63,6 +63,66 @@ def build_snapshot( } +def build_realtime_inputs( + inputs: dict[str, ProviderResult | dict[str, Any]], + directory: dict[str, dict[str, Any]], +) -> dict[str, ProviderResult | dict[str, Any]]: + daily_result = inputs.get("daily") + limits_result = inputs.get("price_limits") + previous_result = inputs.get("previous_limit_up") + if not isinstance(daily_result, ProviderResult) or not isinstance( + limits_result, ProviderResult + ): + return inputs + prices = {str(row.get("ts_code") or ""): row for row in limits_result.rows} + previous = { + str(row.get("ts_code") or ""): row + for row in previous_result.rows + } if isinstance(previous_result, ProviderResult) else {} + pools: dict[str, list[dict[str, Any]]] = { + "limit_up": [], + "limit_down": [], + "broken": [], + } + for quote in daily_result.rows: + identifier = str(quote.get("ts_code") or "") + price = prices.get(identifier) or {} + current = _number(quote.get("close")) + high = _number(quote.get("high")) + up_limit = _number(price.get("up_limit")) + down_limit = _number(price.get("down_limit")) + event_type = "" + if up_limit > 0 and current >= up_limit - 0.001: + event_type = "limit_up" + elif up_limit > 0 and high >= up_limit - 0.001: + event_type = "broken" + elif down_limit > 0 and current <= down_limit + 0.001: + event_type = "limit_down" + if not event_type: + continue + identity = directory.get(identifier) or {} + prior_streak = int(_number(previous.get(identifier, {}).get("limit_times"))) + pools[event_type].append( + { + **quote, + "name": str(quote.get("name") or identity.get("name") or ""), + "industry": str(identity.get("sector") or ""), + "first_time": "", + "last_time": "", + "open_times": 0, + "limit_times": prior_streak + 1 if event_type == "limit_up" else 1, + "fd_amount": 0, + } + ) + metadata = daily_result.metadata + return { + **inputs, + "limit_up": ProviderResult(tuple(pools["limit_up"]), metadata), + "limit_down": ProviderResult(tuple(pools["limit_down"]), metadata), + "broken": ProviderResult(tuple(pools["broken"]), metadata), + } + + def _rows( inputs: dict[str, ProviderResult | dict[str, Any]], key: str ) -> tuple[dict[str, Any], ...]: @@ -74,7 +134,7 @@ def _pool(rows: tuple[dict[str, Any], ...], status: str) -> list[dict[str, Any]] result = [] for row in rows: identifier = str(row.get("ts_code") or "") - amount = _number(row.get("amount")) * 1000 + amount = _amount_yuan(row) result.append( { "identifier": identifier, @@ -238,6 +298,11 @@ def _number(value: Any, default: float = 0.0) -> float: return default +def _amount_yuan(row: dict[str, Any]) -> float: + amount = _number(row.get("amount")) + return amount if row.get("amount_unit") == "yuan" else amount * 1000 + + def _time(value: Any) -> str: text = str(value or "").strip().replace(":", "") if len(text) < 4 or not text[:4].isdigit(): diff --git a/next/backend/features/market/sync.py b/next/backend/features/market/sync.py index 4cfe09e..8132604 100644 --- a/next/backend/features/market/sync.py +++ b/next/backend/features/market/sync.py @@ -11,7 +11,8 @@ from backend.data.providers.base import ProviderError from backend.data.repository import MarketRepository from backend.data.sentiment import calculate_sentiment from backend.database.connection import Database -from backend.features.market.snapshot import build_snapshot +from backend.features.market.events import apply_event_revisions +from backend.features.market.snapshot import build_realtime_inputs, build_snapshot SHANGHAI = ZoneInfo("Asia/Shanghai") @@ -94,6 +95,79 @@ class MarketSnapshotService: "limit_down": len(snapshot["down_limits"]), "broken": len(snapshot["broken"]), "temperature": sentiment["score"], + "source_set": ["tushare", "local"], + "output_version": "market-summary-v1", + } + + def sync_realtime( + self, requested_date: str | None = None, now: datetime | None = None + ) -> dict[str, Any]: + clock = now or datetime.now(SHANGHAI) + target = _date(requested_date or clock.date().isoformat()) + if target != clock.date().isoformat(): + raise SnapshotSyncError("盘中任务只允许同步当前交易日") + local_time = clock.time().replace(tzinfo=None) + in_window = time(9, 15) <= local_time < time(11, 35) or time( + 12, 55 + ) <= local_time < time(15, 5) + if not in_window: + raise SnapshotSyncError("当前不在盘中行情刷新窗口") + with self._database.read() as connection: + dates = self._repository.open_dates(connection, target, 2) + active_count = self._repository.active_stock_count(connection) + if len(dates) < 2 or dates[0] != target: + raise SnapshotSyncError("当前日期不是有效交易日") + if active_count <= 0: + raise SnapshotSyncError("请先同步股票目录") + try: + raw = self._gateway.realtime_snapshot_inputs(target, dates[1]) + except (ProviderError, MarketDataUnavailable) as exc: + raise SnapshotSyncError("盘中行情读取失败，已保留最后成功快照") from exc + daily = raw.get("daily") + price_limits = raw.get("price_limits") + if not isinstance(daily, ProviderResult): + raise SnapshotSyncError("盘中行情缺失，已保留最后成功快照") + coverage = len(daily.rows) / active_count + if coverage < 0.9: + raise SnapshotSyncError( + f"盘中行情覆盖率仅{coverage * 100:.1f}%，已保留最后成功快照" + ) + if not isinstance(price_limits, ProviderResult): + raise SnapshotSyncError("盘中涨跌停价格缺失，已保留最后成功快照") + limit_coverage = len(price_limits.rows) / max(len(daily.rows), 1) + if limit_coverage < 0.95: + raise SnapshotSyncError("盘中涨跌停价格覆盖不足，已保留最后成功快照") + inputs = build_realtime_inputs(raw, self._gateway.stock_directory()) + snapshot = build_snapshot(target, dates[1], inputs) + with self._database.read() as connection: + rows = self._repository.summaries(connection, dates[1], 250) + history = [json.loads(str(row["payload_json"])) for row in rows] + sentiment = calculate_sentiment(snapshot, history) + snapshot["sentiment"] = sentiment + snapshot.update(snapshot["overview"]) + snapshot["temperature"] = sentiment["score"] + observed_at = clock.isoformat(timespec="seconds") + with self._database.transaction() as connection: + self._repository.save_summary( + connection, + trade_date=target, + observed_at=observed_at, + state=SnapshotState.REALTIME.value, + source="tushare", + coverage=min(coverage, 1), + payload=snapshot, + ) + return { + "trade_date": target, + "observed_at": observed_at, + "coverage": round(min(coverage, 1), 4), + "stocks": len(daily.rows), + "limit_up": len(snapshot["limits"]), + "limit_down": len(snapshot["down_limits"]), + "broken": len(snapshot["broken"]), + "temperature": sentiment["score"], + "source_set": ["tushare", "local"], + "output_version": "market-summary-realtime-v1", } def workspace(self, key: str, requested_date: str | None = None) -> dict[str, Any]: @@ -104,6 +178,9 @@ class MarketSnapshotService: if row is None: return {"trade_date": None, "message": "等待管理员首次同步真实收盘行情"} payload = json.loads(str(row["payload_json"])) + with self._database.read() as connection: + revisions = self._repository.event_revisions(connection, str(row["trade_date"])) + apply_event_revisions(payload, revisions) response: dict[str, Any] = { "trade_date": str(row["trade_date"]), "observed_at": str(row["observed_at"]), @@ -173,6 +250,12 @@ class MarketSnapshotService: ).fetchone() factors = json.loads(str(factor_row["payload_json"])) if factor_row else {} snapshot = json.loads(str(summary["payload_json"])) if summary else {} + if summary: + with self._database.read() as connection: + revisions = self._repository.event_revisions( + connection, str(summary["trade_date"]) + ) + apply_event_revisions(snapshot, revisions) event = ( _entity_event(snapshot, entity.identifier, entity.code) if entity_type == "stock" diff --git a/next/backend/features/operations/__init__.py b/next/backend/features/operations/__init__.py new file mode 100644 index 0000000..77f3689 --- /dev/null +++ b/next/backend/features/operations/__init__.py @@ -0,0 +1,3 @@ +from backend.features.operations.service import OperationsService + +__all__ = ["OperationsService"] diff --git a/next/backend/features/operations/routes.py b/next/backend/features/operations/routes.py new file mode 100644 index 0000000..89a56b8 --- /dev/null +++ b/next/backend/features/operations/routes.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +from typing import Annotated + +from fastapi import APIRouter, Path, Query, Request + +from backend.data.gateway import MarketDataUnavailable +from backend.data.providers.base import ProviderError +from backend.features.accounts.auth import ( + AdminPrincipal, + AdminWritePrincipal, + AuthenticatedPrincipal, +) +from backend.features.market.events import MarketEventError +from backend.features.operations.schemas import ( + BackfillInput, + EventRevisionInput, + JobRunResponse, + OperationResult, + RuntimeStatusResponse, +) +from backend.features.operations.service import OperationsError +from backend.http.errors import AppError + +router = APIRouter(tags=["operations"]) + + +@router.get("/operations/status", response_model=RuntimeStatusResponse) +def public_status(request: Request, _principal: AuthenticatedPrincipal) -> dict: + return request.app.state.container.operations.public_status() + + +@router.get("/admin/operations/jobs", response_model=list[JobRunResponse]) +def jobs(request: Request, _principal: AdminPrincipal) -> list[dict]: + return request.app.state.container.operations.jobs() + + +@router.post("/admin/operations/backfill", response_model=OperationResult) +def backfill( + payload: BackfillInput, request: Request, _principal: AdminWritePrincipal +) -> dict: + try: + return request.app.state.container.operations.backfill( + payload.start_date.isoformat(), payload.end_date.isoformat() + ) + except OperationsError as exc: + raise AppError("invalid_operation", str(exc), 400) from exc + + +@router.post("/admin/operations/events/supplement", response_model=OperationResult) +def supplement_events( + request: Request, + _principal: AdminWritePrincipal, + trade_date: Annotated[str, Query(alias="date")], +) -> dict: + try: + return request.app.state.container.operations.supplement_events(trade_date) + except (OperationsError, MarketEventError) as exc: + raise AppError("invalid_operation", str(exc), 400) from exc + except (MarketDataUnavailable, ProviderError) as exc: + raise AppError("event_service_unavailable", str(exc), 503) from exc + + +@router.put( + "/admin/operations/events/{trade_date}/{identifier}", + response_model=OperationResult, +) +def revise_event( + payload: EventRevisionInput, + request: Request, + principal: AdminWritePrincipal, + trade_date: Annotated[str, Path()], + identifier: Annotated[str, Path(min_length=1, max_length=20)], +) -> dict: + try: + return request.app.state.container.operations.revise_event( + principal.user.id, + trade_date=trade_date, + identifier=identifier, + **payload.model_dump(), + ) + except (OperationsError, MarketEventError) as exc: + raise AppError("invalid_event_revision", str(exc), 400) from exc + + +@router.get( + "/admin/operations/events/{trade_date}/{identifier}/history", + response_model=list[OperationResult], +) +def event_history( + request: Request, + _principal: AdminPrincipal, + trade_date: Annotated[str, Path()], + identifier: Annotated[str, Path(min_length=1, max_length=20)], +) -> list[dict]: + return request.app.state.container.operations.event_history(trade_date, identifier) diff --git a/next/backend/features/operations/schemas.py b/next/backend/features/operations/schemas.py new file mode 100644 index 0000000..270b56c --- /dev/null +++ b/next/backend/features/operations/schemas.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from datetime import date +from typing import Any, Literal + +from pydantic import BaseModel, Field + + +class BackfillInput(BaseModel): + start_date: date + end_date: date + + +class EventRevisionInput(BaseModel): + event_type: Literal["limit_up", "broken", "limit_down"] + reason: str = Field(min_length=1, max_length=200) + first_time: str = Field(default="", max_length=5) + last_time: str = Field(default="", max_length=5) + open_times: int | None = Field(default=None, ge=0) + + +class OperationResult(BaseModel): + model_config = {"extra": "allow"} + + +class JobRunResponse(BaseModel): + id: int + kind: str + run_key: str + requested_date: str + trigger: str + status: Literal["running", "completed", "failed"] + attempt: int + started_at: str + finished_at: str | None + duration_ms: int | None + coverage: float | None + source_set: list[str] + output_version: str + payload: dict[str, Any] + error_code: str + error_message: str + + +class RuntimeStatusResponse(BaseModel): + state: Literal["idle", "running", "ready", "degraded"] + message: str + trade_date: str | None = None + observed_at: str | None = None diff --git a/next/backend/features/operations/service.py b/next/backend/features/operations/service.py new file mode 100644 index 0000000..c470f08 --- /dev/null +++ b/next/backend/features/operations/service.py @@ -0,0 +1,312 @@ +from __future__ import annotations + +from datetime import date, datetime, time +from typing import Any +from uuid import uuid4 +from zoneinfo import ZoneInfo + +from backend.data.repository import MarketRepository +from backend.database.connection import Database +from backend.features.market.events import MarketEventService +from backend.features.market.service import MarketService +from backend.features.screener.service import ScreenerService +from backend.jobs.service import JobAlreadyRunning, JobService + +SHANGHAI = ZoneInfo("Asia/Shanghai") + + +class OperationsError(RuntimeError): + pass + + +class OperationsService: + def __init__( + self, + database: Database, + market_repository: MarketRepository, + market: MarketService, + events: MarketEventService, + screener: ScreenerService, + jobs: JobService, + ) -> None: + self._database = database + self._market_repository = market_repository + self._market = market + self._events = events + self._screener = screener + self._jobs = jobs + + def refresh( + self, + requested_date: str | None = None, + *, + trigger: str = "administrator", + realtime: bool = False, + now: datetime | None = None, + run_key: str | None = None, + ) -> dict[str, Any]: + clock = now or datetime.now(SHANGHAI) + target = _date(requested_date or clock.date().isoformat()) + key = run_key or f"{target}:{trigger}:{uuid4().hex}" + try: + return self._jobs.execute( + kind="market.refresh", + run_key=key, + requested_date=target, + trigger=trigger, + operation=lambda: ( + self._market.sync_realtime(target, clock) + if realtime + else self._market.sync_snapshot(target) + ), + stale_after_seconds=120, + ) + except JobAlreadyRunning as exc: + raise OperationsError("行情刷新任务正在运行") from exc + + def backfill(self, start_date: str, end_date: str) -> dict[str, Any]: + start, end = _date(start_date), _date(end_date) + if start > end: + raise OperationsError("开始日期不能晚于结束日期") + with self._database.read() as connection: + dates = self._market_repository.open_dates_between(connection, start, end) + if not dates: + raise OperationsError("所选范围没有交易日") + if len(dates) > 15: + raise OperationsError("单次最多回补15个交易日") + + def operation() -> dict[str, Any]: + rows = [] + for target in dates: + rows.append( + self.refresh( + target, + trigger="historical-backfill", + run_key=f"{target}:archive-v1", + ) + ) + return { + "start_date": start, + "end_date": end, + "dates": list(dates), + "completed": len(rows), + "coverage": min( + (float(row.get("coverage") or 0) for row in rows), default=0 + ), + "source_set": ["tushare", "local"], + "output_version": "historical-backfill-v1", + } + + try: + return self._jobs.execute( + kind="market.backfill", + run_key=f"{start}:{end}", + requested_date=end, + trigger="administrator", + operation=operation, + stale_after_seconds=15 * 60, + ) + except JobAlreadyRunning as exc: + raise OperationsError("历史回补任务正在运行") from exc + + def supplement_events( + self, + trade_date: str, + *, + trigger: str = "administrator", + run_key: str | None = None, + ) -> dict[str, Any]: + target = _date(trade_date) + try: + return self._jobs.execute( + kind="market.event-supplement", + run_key=run_key or f"{target}:event-v1:{uuid4().hex}", + requested_date=target, + trigger=trigger, + operation=lambda: self._events.supplement(target), + stale_after_seconds=120, + ) + except JobAlreadyRunning as exc: + raise OperationsError("事件补充任务正在运行") from exc + + def revise_event(self, user_id: int, **payload: Any) -> dict[str, Any]: + return self._events.revise(user_id=user_id, **payload) + + def event_history(self, trade_date: str, identifier: str) -> list[dict[str, Any]]: + return self._events.history(trade_date, identifier) + + def jobs(self) -> list[dict[str, Any]]: + return self._jobs.latest() + + def public_status(self) -> dict[str, Any]: + latest = self._jobs.latest_for_kind("market.refresh") + with self._database.read() as connection: + snapshot = self._market_repository.latest_summary( + connection, datetime.now(SHANGHAI).date().isoformat() + ) + if latest and latest["status"] == "running": + return {"state": "running", "message": "正在更新行情数据"} + if snapshot is None: + message = "等待管理员首次同步真实行情" + if latest and latest["status"] == "failed": + message = "最近行情更新失败，等待重新同步" + return {"state": "degraded", "message": message} + trade_date = str(snapshot["trade_date"]) + observed_at = str(snapshot["observed_at"]) + if latest and latest["status"] == "failed": + message = "最近更新失败，继续使用最后成功行情" + state = "degraded" + elif str(snapshot["state"]) == "realtime": + message = f"实时行情更新于 {observed_at[11:19]}" + state = "ready" + else: + message = f"{trade_date} 收盘数据已就绪" + state = "ready" + return { + "state": state, + "message": message, + "trade_date": trade_date, + "observed_at": observed_at, + } + + def tick(self, now: datetime | None = None) -> None: + clock = now or datetime.now(SHANGHAI) + target = clock.date().isoformat() + local_time = clock.time().replace(tzinfo=None) + in_market = time(9, 15) <= local_time < time(11, 35) or time( + 12, 55 + ) <= local_time < time(15, 5) + is_open = self._is_open(target) + if ( + in_market + and is_open + and self._jobs.ready_for_schedule( + "market.refresh", + now=clock, + completed_after_seconds=8, + failed_after_seconds=60, + ) + ): + bucket = int(clock.timestamp() // 8) + self._attempt( + lambda: self.refresh( + target, + trigger="realtime-poll", + realtime=True, + now=clock, + run_key=f"{target}:realtime:{bucket}", + ) + ) + if ( + time(9, 15) <= local_time < time(9, 30) + and is_open + and self._jobs.ready_for_schedule( + "auction.collect", + now=clock, + completed_after_seconds=8, + failed_after_seconds=60, + ) + ): + bucket = int(clock.timestamp() // 8) + self._attempt( + lambda: self._jobs.execute( + kind="auction.collect", + run_key=f"{target}:{bucket}", + requested_date=target, + trigger="auction-poll", + operation=lambda: self._auction_result(target, clock), + stale_after_seconds=60, + ) + ) + if local_time >= time(15, 10) and is_open: + final = self._jobs.latest_success("market.refresh", target) + if not final or final["payload"].get("output_version") != "market-summary-v1": + if self._jobs.ready_for_schedule( + "market.refresh", + now=clock, + completed_after_seconds=0, + failed_after_seconds=300, + ): + retry_bucket = int(clock.timestamp() // 300) + self._attempt( + lambda: self.refresh( + target, + trigger="after-close", + run_key=f"{target}:final:{retry_bucket}", + ) + ) + return + self._after_close(target, clock) + + def _after_close(self, target: str, clock: datetime) -> None: + if ( + not self._jobs.latest_success("market.event-supplement", target) + and self._jobs.ready_for_schedule( + "market.event-supplement", + now=clock, + completed_after_seconds=300, + failed_after_seconds=300, + ) + ): + self._attempt( + lambda: self.supplement_events( + target, trigger="after-close", run_key=f"{target}:event-v1" + ) + ) + if ( + not self._jobs.latest_success("screener.after-close", target) + and self._jobs.ready_for_schedule( + "screener.after-close", + now=clock, + completed_after_seconds=300, + failed_after_seconds=300, + ) + ): + self._attempt( + lambda: self._jobs.execute( + kind="screener.after-close", + run_key=f"{target}:catalog-v1", + requested_date=target, + trigger="after-close", + operation=self._screener_result, + stale_after_seconds=15 * 60, + ) + ) + + def _auction_result(self, target: str, clock: datetime) -> dict[str, Any]: + result = self._market.sync_insight("auction", target, 0) + return { + "trade_date": result.get("trade_date") or target, + "coverage": result.get("coverage"), + "phase": result.get("phase"), + "observed_at": result.get("observed_at") or clock.isoformat(timespec="seconds"), + "source_set": ["ifind", "tushare", "local"], + "output_version": "auction-snapshot-v1", + } + + def _screener_result(self) -> dict[str, Any]: + result = self._screener.run_after_close() or {} + return { + **result, + "coverage": result.get("coverage"), + "source_set": ["tushare", "local"], + "output_version": "screener-runs-v1", + } + + def _is_open(self, target: str) -> bool: + with self._database.read() as connection: + return target in self._market_repository.open_dates(connection, target, 1) + + @staticmethod + def _attempt(operation) -> None: + try: + operation() + except Exception: + return + + +def _date(value: str) -> str: + try: + return date.fromisoformat(value).isoformat() + except ValueError as exc: + raise OperationsError("日期格式无效") from exc diff --git a/next/backend/http/router.py b/next/backend/http/router.py index 73fb435..be06780 100644 --- a/next/backend/http/router.py +++ b/next/backend/http/router.py @@ -4,6 +4,7 @@ from backend.features.accounts.routes import router as accounts_router from backend.features.heaven.routes import router as heaven_router from backend.features.market.routes import router as market_router from backend.features.mentor.routes import router as mentor_router +from backend.features.operations.routes import router as operations_router from backend.features.review.routes import router as review_router from backend.features.screener.routes import router as screener_router from backend.http.routes.health import router as health_router @@ -16,3 +17,4 @@ api_router.include_router(screener_router) api_router.include_router(mentor_router) api_router.include_router(heaven_router) api_router.include_router(review_router) +api_router.include_router(operations_router) diff --git a/next/backend/jobs/operations.py b/next/backend/jobs/operations.py new file mode 100644 index 0000000..f145758 --- /dev/null +++ b/next/backend/jobs/operations.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +import asyncio +import logging + +from backend.features.operations.service import OperationsService + +logger = logging.getLogger(__name__) + + +async def run_operations_scheduler( + service: OperationsService, stop: asyncio.Event +) -> None: + while not stop.is_set(): + try: + await asyncio.to_thread(service.tick) + except Exception: + logger.exception( + "Operations scheduler tick failed", + extra={"event": "operations.scheduler.failed"}, + ) + try: + await asyncio.wait_for(stop.wait(), timeout=5) + except TimeoutError: + pass diff --git a/next/backend/jobs/repository.py b/next/backend/jobs/repository.py new file mode 100644 index 0000000..e8a8d74 --- /dev/null +++ b/next/backend/jobs/repository.py @@ -0,0 +1,154 @@ +from __future__ import annotations + +import json +import sqlite3 +from datetime import datetime, timedelta +from typing import Any + + +class JobRepository: + def begin( + self, + connection: sqlite3.Connection, + *, + kind: str, + run_key: str, + requested_date: str, + trigger: str, + started_at: datetime, + stale_after_seconds: int, + ) -> sqlite3.Row: + stale_before = (started_at - timedelta(seconds=stale_after_seconds)).isoformat( + timespec="seconds" + ) + connection.execute( + """ + UPDATE job_runs SET status = 'failed', finished_at = ?, duration_ms = ?, + error_code = 'stale_job', error_message = '任务进程中断或超过最长运行时间' + WHERE kind = ? AND status = 'running' AND started_at < ? + """, + ( + started_at.isoformat(timespec="seconds"), + stale_after_seconds * 1000, + kind, + stale_before, + ), + ) + attempt_row = connection.execute( + """ + SELECT COALESCE(MAX(attempt), 0) + 1 AS attempt + FROM job_runs WHERE kind = ? AND run_key = ? + """, + (kind, run_key), + ).fetchone() + attempt = int(attempt_row["attempt"] if attempt_row else 1) + cursor = connection.execute( + """ + INSERT INTO job_runs ( + kind, run_key, requested_date, trigger, status, attempt, started_at + ) VALUES (?, ?, ?, ?, 'running', ?, ?) + """, + ( + kind, + run_key, + requested_date, + trigger, + attempt, + started_at.isoformat(timespec="seconds"), + ), + ) + return connection.execute( + "SELECT * FROM job_runs WHERE id = ?", (cursor.lastrowid,) + ).fetchone() + + def finish( + self, + connection: sqlite3.Connection, + run_id: int, + *, + finished_at: datetime, + duration_ms: int, + coverage: float | None, + source_set: list[str], + output_version: str, + payload: dict[str, Any], + ) -> None: + connection.execute( + """ + UPDATE job_runs SET status = 'completed', finished_at = ?, duration_ms = ?, + coverage = ?, source_set_json = ?, output_version = ?, payload_json = ? + WHERE id = ? AND status = 'running' + """, + ( + finished_at.isoformat(timespec="seconds"), + duration_ms, + coverage, + _json(source_set), + output_version, + _json(payload), + run_id, + ), + ) + + def fail( + self, + connection: sqlite3.Connection, + run_id: int, + *, + finished_at: datetime, + duration_ms: int, + error_code: str, + error_message: str, + ) -> None: + connection.execute( + """ + UPDATE job_runs SET status = 'failed', finished_at = ?, duration_ms = ?, + error_code = ?, error_message = ? WHERE id = ? AND status = 'running' + """, + ( + finished_at.isoformat(timespec="seconds"), + duration_ms, + error_code, + error_message[:500], + run_id, + ), + ) + + def latest(self, connection: sqlite3.Connection, limit: int = 40) -> tuple[sqlite3.Row, ...]: + return tuple( + connection.execute( + "SELECT * FROM job_runs ORDER BY started_at DESC, id DESC LIMIT ?", + (limit,), + ).fetchall() + ) + + def latest_for_kind( + self, connection: sqlite3.Connection, kind: str + ) -> sqlite3.Row | None: + return connection.execute( + "SELECT * FROM job_runs WHERE kind = ? ORDER BY started_at DESC, id DESC LIMIT 1", + (kind,), + ).fetchone() + + def latest_success( + self, connection: sqlite3.Connection, kind: str, requested_date: str = "" + ) -> sqlite3.Row | None: + return connection.execute( + """ + SELECT * FROM job_runs + WHERE kind = ? AND status = 'completed' AND (? = '' OR requested_date = ?) + ORDER BY finished_at DESC, id DESC LIMIT 1 + """, + (kind, requested_date, requested_date), + ).fetchone() + + +def public_job(row: sqlite3.Row) -> dict[str, Any]: + result = dict(row) + result["source_set"] = json.loads(str(result.pop("source_set_json"))) + result["payload"] = json.loads(str(result.pop("payload_json"))) + return result + + +def _json(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) diff --git a/next/backend/jobs/screener.py b/next/backend/jobs/screener.py deleted file mode 100644 index a428146..0000000 --- a/next/backend/jobs/screener.py +++ /dev/null @@ -1,33 +0,0 @@ -from __future__ import annotations - -import asyncio -import logging - -from backend.features.screener.service import ScreenerError, ScreenerService - -logger = logging.getLogger(__name__) - - -async def run_screener_scheduler(service: ScreenerService, stop: asyncio.Event) -> None: - while not stop.is_set(): - try: - result = await asyncio.to_thread(service.run_after_close) - if result: - logger.info( - "Screener after-close run completed", - extra={"event": "screener.completed", "context": result}, - ) - except ScreenerError as exc: - logger.info( - "Screener is waiting for complete market data", - extra={"event": "screener.waiting", "context": {"reason": str(exc)}}, - ) - except Exception: - logger.exception( - "Screener scheduler failed", - extra={"event": "screener.failed"}, - ) - try: - await asyncio.wait_for(stop.wait(), timeout=300) - except TimeoutError: - pass diff --git a/next/backend/jobs/service.py b/next/backend/jobs/service.py new file mode 100644 index 0000000..295d5fb --- /dev/null +++ b/next/backend/jobs/service.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +import sqlite3 +from collections.abc import Callable +from datetime import datetime +from typing import Any +from zoneinfo import ZoneInfo + +from backend.database.connection import Database +from backend.jobs.repository import JobRepository, public_job + +SHANGHAI = ZoneInfo("Asia/Shanghai") + + +class JobAlreadyRunning(RuntimeError): + pass + + +class JobService: + def __init__(self, database: Database, repository: JobRepository) -> None: + self._database = database + self._repository = repository + + def execute( + self, + *, + kind: str, + run_key: str, + requested_date: str, + trigger: str, + operation: Callable[[], dict[str, Any]], + stale_after_seconds: int, + ) -> dict[str, Any]: + started = datetime.now(SHANGHAI) + try: + with self._database.transaction() as connection: + row = self._repository.begin( + connection, + kind=kind, + run_key=run_key, + requested_date=requested_date, + trigger=trigger, + started_at=started, + stale_after_seconds=stale_after_seconds, + ) + except sqlite3.IntegrityError as exc: + raise JobAlreadyRunning("同类任务正在运行") from exc + run_id = int(row["id"]) + try: + result = operation() + except Exception as exc: + finished = datetime.now(SHANGHAI) + with self._database.transaction() as connection: + self._repository.fail( + connection, + run_id, + finished_at=finished, + duration_ms=_duration(started, finished), + error_code=type(exc).__name__, + error_message=str(exc) or "任务执行失败", + ) + raise + finished = datetime.now(SHANGHAI) + coverage = result.get("coverage") + with self._database.transaction() as connection: + self._repository.finish( + connection, + run_id, + finished_at=finished, + duration_ms=_duration(started, finished), + coverage=float(coverage) if isinstance(coverage, (int, float)) else None, + source_set=[str(value) for value in result.get("source_set") or []], + output_version=str(result.get("output_version") or ""), + payload=result, + ) + return result + + def latest(self, limit: int = 40) -> list[dict[str, Any]]: + with self._database.read() as connection: + return [public_job(row) for row in self._repository.latest(connection, limit)] + + def latest_for_kind(self, kind: str) -> dict[str, Any] | None: + with self._database.read() as connection: + row = self._repository.latest_for_kind(connection, kind) + return public_job(row) if row else None + + def latest_success(self, kind: str, requested_date: str = "") -> dict[str, Any] | None: + with self._database.read() as connection: + row = self._repository.latest_success(connection, kind, requested_date) + return public_job(row) if row else None + + def ready_for_schedule( + self, + kind: str, + *, + now: datetime, + completed_after_seconds: int, + failed_after_seconds: int, + ) -> bool: + """Return whether a scheduler may start another run of this job kind.""" + latest = self.latest_for_kind(kind) + if latest is None: + return True + if latest["status"] == "running": + return False + reference = latest.get("finished_at") or latest.get("started_at") + if not reference: + return True + try: + elapsed = (now - datetime.fromisoformat(str(reference))).total_seconds() + except ValueError: + return True + cooldown = ( + completed_after_seconds + if latest["status"] == "completed" + else failed_after_seconds + ) + return elapsed >= cooldown + + +def _duration(started: datetime, finished: datetime) -> int: + return max(0, round((finished - started).total_seconds() * 1000)) diff --git a/next/docs/evidence/stage-15/acceptance.md b/next/docs/evidence/stage-15/acceptance.md index 581b4ad..ed9a652 100644 --- a/next/docs/evidence/stage-15/acceptance.md +++ b/next/docs/evidence/stage-15/acceptance.md @@ -8,8 +8,8 @@ ## 减法结果 -- 旧运行时代码：55个文件、61,793行；重建运行时代码：206个职责文件、25,151行。 -- 运行时代码净减少36,642行，约59.3%；迁移/备份工具、测试和文档未混入运行时比较。 +- 旧运行时代码：55个文件、61,793行；当前重建运行时代码：216个职责文件、24,581行。 +- 运行时代码净减少37,212行，约60.2%；迁移/备份工具、测试和文档未混入运行时比较。 - 唯一浏览器API、数据网关、LLM网关、弹窗Host、设计令牌和移动规则均通过扫描。 - LLM网关原有3个账户领域具体类型反向导入已改为最小Protocol，未增加第二套服务。 - 超过章程建议行数的算法、Provider和CSS已逐项登记保留原因及拆分触发条件，见`../../final/subtraction-audit.md`。 @@ -17,7 +17,7 @@ ## 维护与回退演练 - 已在真实迁移库副本执行schema 10 -> 9 -> 10，完整性及外键检查通过；明确验证数据回退只能在副本执行。 -- 演练发现并修复`tools.database status`对已有migration误报未知版本的问题；状态查询现在只读，真实副本返回`schema_version=10`。 +- 演练发现并修复`tools.database status`对已有migration误报未知版本的问题；状态查询保持只读；运行治理迁移后当前版本为`schema_version=11`。 - 人工维护入口、常见改动路径、数据源标准、故障定位与复杂度红线见`../../final/maintenance-guide.md`。 - NAS切换、观察和回退动作见`../../final/cutover-checklist.md`；正式容器未变更。 @@ -28,20 +28,20 @@ - 生产响应统一设置CSP、`nosniff`、拒绝Frame、Permissions Policy、Referrer Policy；HTTPS增加HSTS。 - CSP保留Vue动态宽度样式所需的`style-src 'unsafe-inline'`，脚本仍只允许同源。 - 静态哈希资源长期缓存，SPA入口不缓存；未知API不被SPA接管。 -- 真实迁移库副本市场摘要中位数20.90ms、P95 23.55ms；前端生产JS 279.23KB、CSS 89.25KB（未压缩）。 +- 真实迁移库副本市场摘要中位数20.90ms、P95 23.55ms；前端生产JS 296.29KB、CSS 92.08KB（未压缩）。 ## 最终门禁 - Ruff：通过。 -- pytest：99项通过。 +- pytest：104项通过。 - Vue TypeScript：通过。 - Vitest：3个文件、7项通过。 - Vite生产构建：通过。 -- Playwright：17项通过，单worker，最后一轮耗时约1分钟。 +- Playwright：20项通过，单worker，最后一轮耗时约1.7分钟。 - `git diff --check`和已知敏感值扫描：通过。 - Docker：本机未安装，未虚报实构建；列为NAS切换前阻断项。 ## 结论 -从零重建、数据迁移工具、备份恢复、结构治理、移动端、性能安全和人工维护资产已经完成。 -当前代码具备进入独立NAS预发布验证的条件，但未经用户最终确认不会替换正式容器。 +模块化重建基线、数据迁移工具、备份恢复、运行治理、移动端基线、性能安全和人工维护资产已经建立。 +产品仍有`../../final/completion-audit.md`登记的明确缺口，尚不能宣称完整迁移；未经用户最终确认也不会替换正式容器。 diff --git a/next/docs/final/completion-audit.md b/next/docs/final/completion-audit.md index 9f0357d..ee3aaf8 100644 --- a/next/docs/final/completion-audit.md +++ b/next/docs/final/completion-audit.md @@ -25,22 +25,10 @@ ### P0：用户可见功能缺失 -1. **历史数据回补**：系统管理仍显示“暂不可用”，不符合行情管理的正式功能要求。 -2. **模型连通性测试**：模型池支持增删改和主辅选择，但每个模型的独立测试入口与服务端测试调用尚未实现。 -3. **事件原因治理**：涨停、炸板、跌停原因的盘后补充、管理员人工修订优先级和修订记录尚未实现。 -4. **自定义选股完整能力**：缺少自然语言转换受控公式和滚动回测；精选策略详情尚未完整显示评分权重、 +1. **模型连通性测试**：模型池支持增删改和主辅选择，但每个模型的独立测试入口与服务端测试调用尚未实现。 +2. **自定义选股完整能力**：缺少自然语言转换受控公式和滚动回测；精选策略详情尚未完整显示评分权重、 执行频率和风险等级。 -### P0：运行与数据真相缺失 - -1. **盘中行情刷新任务**：当前只启动盘后选股调度。交易日 9:15-11:35、12:55-15:05 的行情刷新、 - 约 8 秒节流、单任务互斥和最后成功快照保护尚未形成正式后台任务。 -2. **任务状态与审计**：行情刷新、竞价采集、盘后选股和事件补充没有统一的开始/完成/失败、覆盖率、 - 输出版本、耗时和重试记录。前端也无法显示真实任务状态。 -3. **阶段选股四步状态**：已去掉“无运行结果时回退第一条结果”的错误行为，但四步仍未逐步表达未执行、 - 执行中、失败和完成。 -4. **状态栏真相**：状态栏右侧仍是固定“等待行情数据”，没有反映真实快照或后台任务状态。 - ### P1：验收和交互覆盖不足 1. 固定案例 `C06`、`C08` 当前为未实现，不能标记通过。 @@ -63,6 +51,14 @@ - 个股详情已包含交易指标、资金流、事件逻辑、自选操作、观势入口和私有个股笔记；题材详情展示成分股。 - 股池、天梯、轮动、题材、人气榜、龙虎榜、智能选股、自选、交易日志、策略跟踪和提醒中的股票代码 共用唯一日K/分时悬浮与详情入口；无悬停设备直接进入详情。 +- 系统管理已提供最多15个交易日的历史回补入口；任务结果和失败原因进入统一任务审计。 +- 盘中行情刷新、动态竞价、盘后正式刷新、事件补充和盘后选股共用唯一调度器与任务表；同类任务互斥， + 成功与失败按任务类型统一冷却，外部接口故障不会每5秒重复写失败记录。 +- 盘中正式计算仅使用Tushare全市场实时行情和官方涨跌停价格；覆盖率不达门槛时拒绝写入并保留最后成功快照。 +- 涨停、炸板、跌停原因可由iFinD盘后补充，管理员修订具有更高优先级，全部修订历史永久保留。 +- 阶段选股四步和候选区已直接表达未执行、执行中、失败、已完成；策略库不再把未执行或失败误写为暂无信号。 +- 状态栏读取真实后台任务与最后成功行情状态，不再显示固定占位文字。 +- 本轮验证为Ruff、104项pytest、Vue类型检查、7项Vitest、生产构建和20项Playwright全部通过。 ## 外部环境阻断项 diff --git a/next/frontend/src/app/shell/StatusBar.vue b/next/frontend/src/app/shell/StatusBar.vue index 29d1255..bbc7f06 100644 --- a/next/frontend/src/app/shell/StatusBar.vue +++ b/next/frontend/src/app/shell/StatusBar.vue @@ -1,7 +1,8 @@ diff --git a/next/frontend/src/app/system/MarketSettingsPanel.vue b/next/frontend/src/app/system/MarketSettingsPanel.vue index 3185089..21b4d24 100644 --- a/next/frontend/src/app/system/MarketSettingsPanel.vue +++ b/next/frontend/src/app/system/MarketSettingsPanel.vue @@ -2,6 +2,8 @@ import { onMounted, reactive, ref } from "vue"; import { api } from "../../shared/api/client"; +import { operationsApi, type JobRun } from "../../shared/api/operations"; +import { useMarketStore } from "../../shared/stores/market"; import { useUiStore } from "../../shared/stores/ui"; type CredentialStatus = { name: string; configured: boolean; updated_at: string | null }; @@ -12,6 +14,7 @@ const labels: Record = { ifind_access_token: "iFinD Access Token", }; const ui = useUiStore(); +const market = useMarketStore(); const statuses = ref([]); const values = reactive>({}); const loading = ref(true); @@ -19,11 +22,36 @@ const errorMessage = ref(""); const syncing = ref(false); const syncResult = ref(""); const syncError = ref(""); +const jobs = ref([]); +const backfill = reactive({ start_date: market.selectedDate, end_date: market.selectedDate }); +const backfilling = ref(false); +const operationMessage = ref(""); +const event = reactive({ + trade_date: market.selectedDate, + identifier: "", + event_type: "limit_up" as "limit_up" | "broken" | "limit_down", + reason: "", + first_time: "", + last_time: "", + open_times: null as number | null, +}); +const eventBusy = ref(false); +const eventHistory = ref[]>([]); + +const jobLabels: Record = { + "market.refresh": "行情刷新", + "market.backfill": "历史回补", + "auction.collect": "竞价采集", + "market.event-supplement": "事件补充", + "screener.after-close": "盘后选股", +}; +const statusLabels = { running: "执行中", completed: "已完成", failed: "失败" }; async function load(): Promise { loading.value = true; try { statuses.value = await api.get("/admin/system/credentials"); + jobs.value = await operationsApi.jobs(); } catch (error) { errorMessage.value = error instanceof Error ? error.message : "行情凭据读取失败。"; } finally { @@ -31,6 +59,70 @@ async function load(): Promise { } } +async function loadJobs(): Promise { + try { jobs.value = await operationsApi.jobs(); } + catch (error) { ui.showToast(error instanceof Error ? error.message : "任务状态读取失败"); } +} + +async function runBackfill(): Promise { + backfilling.value = true; + operationMessage.value = ""; + try { + const result = await operationsApi.backfill(backfill.start_date, backfill.end_date); + operationMessage.value = `已完成 ${Number(result.completed ?? 0)} 个交易日回补`; + ui.showToast("历史数据回补已完成"); + await loadJobs(); + } catch (error) { + ui.showToast(error instanceof Error ? error.message : "历史回补失败"); + } finally { backfilling.value = false; } +} + +async function supplementEvents(): Promise { + eventBusy.value = true; + try { + const result = await operationsApi.supplementEvents(event.trade_date); + operationMessage.value = `事件原因已匹配 ${Number(result.matched ?? 0)} 条`; + ui.showToast("事件原因补充完成"); + await loadJobs(); + } catch (error) { + ui.showToast(error instanceof Error ? error.message : "事件补充暂不可用"); + } finally { eventBusy.value = false; } +} + +async function reviseEvent(): Promise { + eventBusy.value = true; + try { + await operationsApi.reviseEvent(event.trade_date, event.identifier.trim(), { + event_type: event.event_type, + reason: event.reason, + first_time: event.first_time, + last_time: event.last_time, + open_times: event.open_times, + }); + operationMessage.value = "人工修订已保存，并优先于自动补充内容"; + ui.showToast("事件原因修订已保存"); + await showEventHistory(); + } catch (error) { + ui.showToast(error instanceof Error ? error.message : "事件修订失败"); + } finally { eventBusy.value = false; } +} + +async function showEventHistory(): Promise { + if (!event.identifier.trim()) return; + try { + eventHistory.value = await operationsApi.eventHistory( + event.trade_date, event.identifier.trim(), + ); + } catch (error) { + ui.showToast(error instanceof Error ? error.message : "修订记录读取失败"); + } +} + +function duration(value: number | null): string { + if (value === null) return ""; + return value < 1000 ? `${value} ms` : `${(value / 1000).toFixed(1)} s`; +} + async function save(name: string): Promise { const value = values[name]?.trim(); if (!value) { @@ -97,11 +189,32 @@ onMounted(load);
-

历史数据回补

暂不可用
-
-

历史数据回补将在行情数据迁移完成后开放。

- -
+

历史数据回补

单次最多15个交易日
+
+ + + +
+
+
+

事件原因治理

人工修订优先并保留记录
+
+ + + + + + + +
+
+
时间来源类型原因修订人
{{ String(row.created_at).replace('T', ' ').slice(0, 19) }}{{ row.source === 'admin' ? '人工修订' : '自动补充' }}{{ row.event_type }}{{ row.reason }}{{ row.created_by_name ?? '' }}
+
+
+

后台任务状态

+

{{ operationMessage }}

+
任务数据日期状态开始时间耗时覆盖率（%）说明
{{ jobLabels[job.kind] ?? job.kind }}{{ job.requested_date }}{{ statusLabels[job.status] }}{{ job.started_at.replace('T', ' ').slice(0, 19) }}{{ duration(job.duration_ms) }}{{ job.coverage === null ? '' : (job.coverage * 100).toFixed(1) }}{{ job.error_message }}
+

暂无后台任务记录

diff --git a/next/frontend/src/pages/screener/CandidateTable.vue b/next/frontend/src/pages/screener/CandidateTable.vue index 5687e6a..be3f217 100644 --- a/next/frontend/src/pages/screener/CandidateTable.vue +++ b/next/frontend/src/pages/screener/CandidateTable.vue @@ -19,13 +19,21 @@ function number(value: number | null, digits = 2): string { {{ run.items.length }} 只 -
+
+ 盘后选股尚未执行 + 当日收盘行情定稿后由后台自动计算并归档。 +
+
+ 正在计算候选结果 + 完成后会自动保存本次结果。 +
+
数据尚不完整：{{ run.missing_fields.join("、") }}
-
- 本次计算失败，已隔离该策略，不影响其他策略。 +
+ 本次计算失败，已隔离该策略，不影响其他策略。{{ run.error_message }}
-
+
暂无符合条件个股 完整数据下无信号会保留为空，不补造候选。
diff --git a/next/frontend/src/pages/screener/StagePanel.vue b/next/frontend/src/pages/screener/StagePanel.vue index eedf354..14b6bc1 100644 --- a/next/frontend/src/pages/screener/StagePanel.vue +++ b/next/frontend/src/pages/screener/StagePanel.vue @@ -9,7 +9,18 @@ const emit = defineEmits<{ track: [run: ScreenerRun, candidate: Candidate] }>(); const selected = ref(props.runs[0]?.strategy_id ?? props.strategies[0]?.id ?? ""); const run = computed(() => props.runs.find((item) => item.strategy_id === selected.value)); const strategy = computed(() => props.strategies.find((item) => item.id === (run.value?.strategy_id ?? selected.value))); -const hasRun = computed(() => Boolean(run.value)); +const runState = computed<"idle" | "running" | "failed" | "completed">(() => { + if (!run.value || run.value.status === "pending") return "idle"; + if (run.value.status === "running") return "running"; + if (run.value.status === "failed") return "failed"; + return "completed"; +}); +const runStateLabel = computed(() => ({ + idle: "未执行", + running: "执行中", + failed: "失败", + completed: "已完成", +})[runState.value]); watch( () => props.runs, @@ -26,10 +37,13 @@ watch(
当前阶段自动候选{{ strategy?.display_name ?? "等待盘后判定" }}

{{ strategy?.description ?? "每日收盘数据定稿后自动生成，结果允许为空。" }}

- 盘后自动 + {{ runStateLabel }}
- 阶段识别策略匹配自动计算结果归档 + 阶段识别 + 策略匹配 + 自动计算 + 结果归档