from __future__ import annotations import csv import hashlib import io import json import math import os import tempfile from dataclasses import asdict, dataclass from datetime import date, datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any, Iterable, Sequence import mysql.connector from .database import MySQLSettings, connect_target, safe_mysql_error SCHEMA_VERSION = 1 SHA256_LENGTH = 64 V1_POSITION_LABELS = { "BELOW_BASE_RANGE": "偏低", "WITHIN_BASE_RANGE": "基本合理", "ABOVE_BASE_RANGE": "偏贵", "ABOVE_OPTIMISTIC_RANGE": "明显偏贵", } class LedgerError(RuntimeError): def __init__(self, code: str, message: str): self.code = code super().__init__(message) @dataclass(frozen=True) class ValuationCandidate: ticker: str company: str market: str currency: str valuation_date: str method: str pessimistic_low: float pessimistic_high: float base_low: float base_high: float optimistic_low: float optimistic_high: float normalized_profit: float | None normalized_pe: float | None pb: float | None ps: float | None consensus_year: int | None consensus_profit: float | None consensus_count: int | None report_path: str snapshot_path: str source_hash: str priority: int = 1 @property def valuation_id(self) -> str: raw = f"{self.ticker}|{self.valuation_date}|{self.source_hash}".encode("utf-8") return "VAL-" + hashlib.sha256(raw).hexdigest()[:24] @dataclass(frozen=True) class PriceRecord: ticker: str trade_date: str close: float currency: str source_id: str source_timestamp: str is_complete_trading_day: bool = True def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def atomic_write(path: Path, data: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) descriptor, raw_temp = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) temp = Path(raw_temp) try: with os.fdopen(descriptor, "wb") as handle: handle.write(data) handle.flush() os.fsync(handle.fileno()) os.replace(temp, path) finally: if temp.exists(): temp.unlink() def _load_json(path: Path) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8-sig")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise LedgerError("E_INPUT_JSON", f"无法读取 JSON：{path}: {exc}") from exc if not isinstance(value, dict): raise LedgerError("E_INPUT_SCHEMA", f"JSON 根节点必须为对象：{path}") return value def _finite_float(value: Any, field: str, *, optional: bool = False) -> float | None: if value is None and optional: return None try: number = Decimal(str(value)) except (InvalidOperation, TypeError, ValueError) as exc: raise LedgerError("E_INPUT_SCHEMA", f"{field} 必须是数字") from exc if not number.is_finite(): raise LedgerError("E_INPUT_SCHEMA", f"{field} 必须是有限数字") return float(number) def _iso_date(value: Any, field: str) -> str: if not isinstance(value, str): raise LedgerError("E_INPUT_SCHEMA", f"{field} 必须是 YYYY-MM-DD") try: return date.fromisoformat(value[:10]).isoformat() except ValueError as exc: raise LedgerError("E_INPUT_SCHEMA", f"{field} 不是有效日期：{value}") from exc def _canonical_timestamp(value: Any, field: str) -> str: if not isinstance(value, str): raise LedgerError("E_INPUT_SCHEMA", f"{field} 必须是带时区 ISO-8601") try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise LedgerError("E_INPUT_SCHEMA", f"{field} 不是有效时间：{value}") from exc if parsed.tzinfo is None: raise LedgerError("E_INPUT_SCHEMA", f"{field} 必须带时区") return parsed.isoformat() def _relative(path: Path, project_root: Path) -> str: try: return path.resolve(strict=True).relative_to(project_root.resolve(strict=True)).as_posix() except (OSError, ValueError) as exc: raise LedgerError("E_PATH_SCOPE", f"正式文件不在项目根目录内：{path}") from exc def _scenario(results: dict[str, Any], role: str) -> dict[str, Any]: rows = results.get("scenarios") if not isinstance(rows, list): raise LedgerError("E_INPUT_SCHEMA", "valuation_results.scenarios 必须是数组") matches = [row for row in rows if isinstance(row, dict) and row.get("role") == role] if len(matches) != 1: raise LedgerError("E_INPUT_SCHEMA", f"估值结果必须且只能有一个 {role} 情景") return matches[0] def _normalized_market(ticker: str, raw_market: Any) -> str: if not isinstance(raw_market, str) or not raw_market.strip(): raise LedgerError("E_INPUT_SCHEMA", "meta.market 缺失") suffixes = { ".SH": "上交所", ".SZ": "深交所", ".BJ": "北交所", ".HK": "港交所", } for suffix, market in suffixes.items(): if ticker.endswith(suffix): return market raise LedgerError("E_INPUT_SCHEMA", f"证券代码市场后缀不受支持：{ticker}") def _candidate_from_package( snapshot_path: Path, results_path: Path, report_path: Path, project_root: Path, *, priority: int, batch_row: dict[str, Any] | None = None, ) -> ValuationCandidate: snapshot = _load_json(snapshot_path) results = _load_json(results_path) snapshot_meta = snapshot.get("meta") result_meta = results.get("meta") if not isinstance(snapshot_meta, dict) or not isinstance(result_meta, dict): raise LedgerError("E_INPUT_SCHEMA", "快照和计算结果都必须包含 meta") fields = ("code", "company", "as_of_date", "currency") for field in fields: if snapshot_meta.get(field) != result_meta.get(field): raise LedgerError("E_INPUT_MISMATCH", f"snapshot/results meta.{field} 不一致") ticker = str(result_meta.get("code") or "").upper() if "." not in ticker: raise LedgerError("E_INPUT_SCHEMA", f"证券代码缺市场后缀：{ticker}") currency = str(result_meta.get("currency") or "") if currency not in {"CNY", "HKD"}: raise LedgerError("E_INPUT_SCHEMA", f"不支持的币种：{currency}") pessimistic = _scenario(results, "pessimistic") base = _scenario(results, "base") optimistic = _scenario(results, "optimistic") ranges = { "pessimistic_low": _finite_float(pessimistic.get("price_low"), "pessimistic.price_low"), "pessimistic_high": _finite_float(pessimistic.get("price_high"), "pessimistic.price_high"), "base_low": _finite_float(base.get("price_low"), "base.price_low"), "base_high": _finite_float(base.get("price_high"), "base.price_high"), "optimistic_low": _finite_float(optimistic.get("price_low"), "optimistic.price_low"), "optimistic_high": _finite_float(optimistic.get("price_high"), "optimistic.price_high"), } if ranges["pessimistic_low"] > ranges["pessimistic_high"]: raise LedgerError("E_RANGE", "悲观区间上下沿反转") if ranges["base_low"] > ranges["base_high"]: raise LedgerError("E_RANGE", "基准区间上下沿反转") if ranges["optimistic_low"] > ranges["optimistic_high"]: raise LedgerError("E_RANGE", "乐观区间上下沿反转") if ranges["base_high"] > ranges["optimistic_high"]: raise LedgerError("E_RANGE", "基准上沿不得高于乐观上沿") metrics = results.get("metrics") if isinstance(results.get("metrics"), dict) else {} institution_summary = ( results.get("institutions", {}).get("summary", {}) if isinstance(results.get("institutions"), dict) else {} ) consensus_year: int | None = None consensus_profit: float | None = None consensus_count: int | None = None numeric_years = sorted( int(year) for year in institution_summary if str(year).isdigit() ) if numeric_years: consensus_year = numeric_years[0] summary = institution_summary.get(str(consensus_year), {}) if isinstance(summary, dict): consensus_profit = _finite_float(summary.get("profit_mean"), "institutions.profit_mean", optional=True) raw_count = summary.get("institution_count") consensus_count = int(raw_count) if raw_count is not None else None analysis = snapshot.get("analysis") if isinstance(snapshot.get("analysis"), dict) else {} if analysis.get("consensus_org_count") is not None: consensus_count = int(analysis["consensus_org_count"]) if batch_row: if batch_row.get("consensus_2026") is not None: consensus_year = 2026 consensus_profit = _finite_float(batch_row["consensus_2026"], "batch.consensus_2026") if batch_row.get("consensus_count") is not None: consensus_count = int(batch_row["consensus_count"]) method = str(base.get("method") or "").upper() if method not in {"PE", "PB", "PS"}: raise LedgerError("E_INPUT_SCHEMA", f"不支持的基准估值方法：{method}") return ValuationCandidate( ticker=ticker, company=str(result_meta.get("company") or ""), market=_normalized_market(ticker, result_meta.get("market")), currency=currency, valuation_date=_iso_date(result_meta.get("as_of_date"), "meta.as_of_date"), method=method, normalized_profit=_finite_float(metrics.get("normalized_profit"), "metrics.normalized_profit", optional=True), normalized_pe=_finite_float(metrics.get("normalized_pe"), "metrics.normalized_pe", optional=True), pb=_finite_float(metrics.get("pb"), "metrics.pb", optional=True), ps=_finite_float(metrics.get("ps"), "metrics.ps", optional=True), consensus_year=consensus_year, consensus_profit=consensus_profit, consensus_count=consensus_count, report_path=_relative(report_path, project_root), snapshot_path=_relative(snapshot_path, project_root), source_hash=sha256_file(snapshot_path), priority=priority, **ranges, ) def _resolve_reference(batch_path: Path, relative: str, project_root: Path) -> Path: raw = Path(relative) if raw.is_absolute() or ".." in raw.parts: raise LedgerError("E_PATH_SCOPE", f"批次引用必须是安全相对路径：{relative}") project = project_root.resolve(strict=True) for parent in (batch_path.parent, *batch_path.parents): try: parent.resolve(strict=True).relative_to(project) except (OSError, ValueError): continue candidate = parent / raw if candidate.is_file(): return candidate if parent.resolve() == project: break raise LedgerError("E_INPUT_MISSING", f"批次正式报告不存在：{relative}") def _candidate_from_batch_row( batch_path: Path, row: dict[str, Any], project_root: Path ) -> ValuationCandidate: formal_path = row.get("formal_path") if not isinstance(formal_path, str) or not formal_path: raise LedgerError("E_INPUT_SCHEMA", "batch row 缺少 formal_path") report = _resolve_reference(batch_path, formal_path, project_root) snapshots = sorted(report.parent.glob("*估值快照_*.json")) results = report.parent / "calculation" / "valuation_results.json" if len(snapshots) != 1 or not results.is_file(): raise LedgerError("E_INPUT_MISSING", f"正式包缺少唯一快照或 calculation：{report.parent}") candidate = _candidate_from_package( snapshots[0], results, report, project_root, priority=3, batch_row=row ) if row.get("ticker") != candidate.ticker or row.get("company") != candidate.company: raise LedgerError("E_INPUT_MISMATCH", f"批次索引与正式包身份不一致：{formal_path}") return candidate def discover_valuations( results_root: Path, project_root: Path ) -> tuple[list[ValuationCandidate], list[dict[str, str]]]: if not results_root.is_dir(): raise LedgerError("E_INPUT_MISSING", f"估值结果目录不存在：{results_root}") candidates: list[ValuationCandidate] = [] gaps: list[dict[str, str]] = [] for batch_path in sorted(results_root.rglob("batch_results_83.json")): try: payload = _load_json(batch_path) rows = payload.get("rows") if not isinstance(rows, list): raise LedgerError("E_INPUT_SCHEMA", "batch_results_83.rows 必须是数组") for index, row in enumerate(rows): try: if not isinstance(row, dict): raise LedgerError("E_INPUT_SCHEMA", "batch row 必须是对象") candidates.append(_candidate_from_batch_row(batch_path, row, project_root)) except LedgerError as exc: gaps.append({"path": f"{batch_path}#rows[{index}]", "code": exc.code, "message": str(exc)}) except LedgerError as exc: gaps.append({"path": str(batch_path), "code": exc.code, "message": str(exc)}) for snapshot_path in sorted(results_root.rglob("*估值快照_*.json")): results_path = snapshot_path.parent / "calculation" / "valuation_results.json" reports = sorted(snapshot_path.parent.glob("*价格合理性评估_*.md")) try: if not results_path.is_file() or len(reports) != 1: raise LedgerError("E_INPUT_MISSING", "正式包缺少 calculation/valuation_results.json 或唯一正式报告") candidates.append( _candidate_from_package( snapshot_path, results_path, reports[0], project_root, priority=1 ) ) except LedgerError as exc: gaps.append({"path": str(snapshot_path), "code": exc.code, "message": str(exc)}) by_date: dict[tuple[str, str], list[ValuationCandidate]] = {} for candidate in candidates: by_date.setdefault((candidate.ticker, candidate.valuation_date), []).append(candidate) selected: list[ValuationCandidate] = [] for key, rows in sorted(by_date.items()): unique = {row.source_hash: row for row in rows} winner = max(unique.values(), key=lambda row: (row.priority, row.snapshot_path, row.source_hash)) selected.append(winner) for source_hash, row in unique.items(): if source_hash != winner.source_hash: gaps.append( { "path": row.snapshot_path, "code": "DUPLICATE_VALUATION_DATE_SKIPPED", "message": f"{key[0]} {key[1]} 已选择更高优先级正式版本 {winner.snapshot_path}", } ) if not selected: raise LedgerError("E_NO_VALUATIONS", "没有发现可结构化导入的正式估值版本") return sorted(selected, key=lambda row: (row.ticker, row.valuation_date)), gaps EXPECTED_TABLE_COLUMNS = { "security": {"ticker", "company", "market", "currency", "active", "created_at", "updated_at"}, "valuation_version": { "valuation_id", "ticker", "valuation_date", "method", "pessimistic_low", "pessimistic_high", "base_low", "base_high", "optimistic_low", "optimistic_high", "normalized_profit", "normalized_pe", "pb", "ps", "consensus_year", "consensus_profit", "consensus_count", "report_path", "snapshot_path", "source_hash", "active_from", "active_to", "created_at", }, "daily_price": { "ticker", "trade_date", "close", "currency", "source_id", "source_timestamp", "ingested_at", }, "daily_judgement": { "ticker", "trade_date", "valuation_id", "close", "base_low", "base_high", "optimistic_high", "label", "distance_to_base_low", "distance_to_base_high", "computed_at", }, } def connect_database(settings: MySQLSettings): try: return connect_target(settings) except mysql.connector.Error as exc: raise LedgerError("E_DATABASE", safe_mysql_error(exc)) from exc def _schema_statements() -> list[str]: schema = Path(__file__).with_name("schema.sql").read_text(encoding="utf-8") return [part.strip() for part in schema.split(";") if part.strip()] def _validate_schema(connection, database: str) -> None: cursor = connection.cursor(dictionary=True) try: cursor.execute( "SELECT TABLE_NAME, ENGINE FROM information_schema.TABLES " "WHERE TABLE_SCHEMA=%s AND TABLE_TYPE='BASE TABLE'", (database,), ) tables = {row["TABLE_NAME"]: row["ENGINE"] for row in cursor.fetchall()} if set(tables) != set(EXPECTED_TABLE_COLUMNS): raise LedgerError("E_SCHEMA", f"业务表集合不正确：{sorted(tables)}") if any(engine != "InnoDB" for engine in tables.values()): raise LedgerError("E_SCHEMA", "四张业务表必须全部使用 InnoDB") cursor.execute( "SELECT TABLE_NAME, COLUMN_NAME FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA=%s", (database,), ) actual = {name: set() for name in EXPECTED_TABLE_COLUMNS} for row in cursor.fetchall(): if row["TABLE_NAME"] in actual: actual[row["TABLE_NAME"]].add(row["COLUMN_NAME"]) if actual != EXPECTED_TABLE_COLUMNS: raise LedgerError("E_SCHEMA", "业务表字段合同与 schema.sql 不一致") finally: cursor.close() def initialize_database(settings: MySQLSettings) -> None: connection = connect_database(settings) cursor = connection.cursor() try: for statement in _schema_statements(): cursor.execute(statement) connection.commit() _validate_schema(connection, settings.database) except mysql.connector.Error as exc: connection.rollback() raise LedgerError("E_DATABASE", safe_mysql_error(exc)) from exc except BaseException: connection.rollback() raise finally: cursor.close() connection.close() def _write_gaps(path: Path, gaps: Sequence[dict[str, str]]) -> None: buffer = io.StringIO(newline="") writer = csv.DictWriter(buffer, fieldnames=["path", "code", "message"], lineterminator="\n") writer.writeheader() writer.writerows(gaps) atomic_write(path, buffer.getvalue().encode("utf-8")) def import_valuations( settings: MySQLSettings, results_root: Path, project_root: Path, gaps_path: Path, ) -> dict[str, Any]: initialize_database(settings) candidates, gaps = discover_valuations(results_root, project_root) connection = connect_database(settings) cursor = connection.cursor(dictionary=True) inserted = 0 unchanged = 0 affected: set[str] = set() now = datetime.now(timezone.utc).replace(tzinfo=None) latest_security: dict[str, ValuationCandidate] = {} for candidate in candidates: current = latest_security.get(candidate.ticker) if current is None or ( candidate.valuation_date, candidate.priority, candidate.source_hash ) > (current.valuation_date, current.priority, current.source_hash): latest_security[candidate.ticker] = candidate try: connection.start_transaction() for row in candidates: cursor.execute( "SELECT company, market, currency FROM security WHERE ticker=%s", (row.ticker,), ) existing_security = cursor.fetchone() security_row = latest_security[row.ticker] if existing_security and ( existing_security["market"], existing_security["currency"] ) != (security_row.market, security_row.currency): gaps.append( { "path": row.snapshot_path, "code": "E_SECURITY_MISMATCH", "message": f"{row.ticker} 证券身份与数据库不一致", } ) continue if existing_security is None: cursor.execute( "INSERT INTO security(ticker, company, market, currency, active, created_at, updated_at) " "VALUES (%s, %s, %s, %s, 1, %s, %s)", ( security_row.ticker, security_row.company, security_row.market, security_row.currency, now, now, ), ) elif existing_security["company"] != security_row.company: cursor.execute( "UPDATE security SET company=%s, updated_at=%s WHERE ticker=%s", (security_row.company, now, security_row.ticker), ) cursor.execute( "SELECT * FROM valuation_version WHERE ticker=%s AND valuation_date=%s", (row.ticker, row.valuation_date), ) same_date = cursor.fetchall() if same_date and all(item["source_hash"] != row.source_hash for item in same_date): gaps.append( { "path": row.snapshot_path, "code": "E_DUPLICATE_VALUATION_DATE", "message": f"{row.ticker} {row.valuation_date} 已有不同正式版本，拒绝产生重叠有效期", } ) continue matching = next( (item for item in same_date if item["source_hash"] == row.source_hash), None, ) if matching is not None: expected = { "valuation_id": row.valuation_id, "ticker": row.ticker, "valuation_date": row.valuation_date, "method": row.method, "pessimistic_low": row.pessimistic_low, "pessimistic_high": row.pessimistic_high, "base_low": row.base_low, "base_high": row.base_high, "optimistic_low": row.optimistic_low, "optimistic_high": row.optimistic_high, "normalized_profit": row.normalized_profit, "normalized_pe": row.normalized_pe, "pb": row.pb, "ps": row.ps, "consensus_year": row.consensus_year, "consensus_profit": row.consensus_profit, "consensus_count": row.consensus_count, "report_path": row.report_path, "snapshot_path": row.snapshot_path, "source_hash": row.source_hash, } mismatches = [ name for name, value in expected.items() if not _immutable_equal(matching.get(name), value) ] if mismatches: gaps.append( { "path": row.snapshot_path, "code": "E_IMMUTABLE_VERSION_CONFLICT", "message": "同一估值身份的不可变字段冲突：" + ",".join(mismatches), } ) else: unchanged += 1 continue values = ( row.valuation_id, row.ticker, row.valuation_date, row.method, row.pessimistic_low, row.pessimistic_high, row.base_low, row.base_high, row.optimistic_low, row.optimistic_high, row.normalized_profit, row.normalized_pe, row.pb, row.ps, row.consensus_year, row.consensus_profit, row.consensus_count, row.report_path, row.snapshot_path, row.source_hash, row.valuation_date, now, ) cursor.execute( """ INSERT INTO valuation_version( valuation_id, ticker, valuation_date, method, pessimistic_low, pessimistic_high, base_low, base_high, optimistic_low, optimistic_high, normalized_profit, normalized_pe, pb, ps, consensus_year, consensus_profit, consensus_count, report_path, snapshot_path, source_hash, active_from, active_to, created_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NULL, %s) """, values, ) inserted += 1 affected.add(row.ticker) for ticker in affected: cursor.execute( "SELECT valuation_id, valuation_date FROM valuation_version " "WHERE ticker=%s ORDER BY valuation_date, valuation_id", (ticker,), ) versions = cursor.fetchall() for index, version in enumerate(versions): valuation_date = _date_value(version["valuation_date"]) active_from = valuation_date if index == 0 else valuation_date + timedelta(days=1) active_to = None if index + 1 < len(versions): active_to = _date_value(versions[index + 1]["valuation_date"]) cursor.execute( "UPDATE valuation_version SET active_from=%s, active_to=%s " "WHERE valuation_id=%s", (active_from, active_to, version["valuation_id"]), ) connection.commit() except mysql.connector.Error as exc: connection.rollback() raise LedgerError("E_DATABASE", safe_mysql_error(exc)) from exc except BaseException: connection.rollback() raise finally: cursor.close() connection.close() _write_gaps(gaps_path, gaps) return { "status": "IMPORTED", "discovered": len(candidates), "inserted": inserted, "unchanged": unchanged, "gap_count": len(gaps), "gaps_path": str(gaps_path), } def _date_value(value: Any) -> date: if isinstance(value, datetime): return value.date() if isinstance(value, date): return value return date.fromisoformat(str(value)[:10]) def _immutable_equal(actual: Any, expected: Any) -> bool: if isinstance(actual, (date, datetime)): actual = actual.date().isoformat() if isinstance(actual, datetime) else actual.isoformat() if expected is None or actual is None: return actual is None and expected is None if isinstance(actual, (Decimal, float, int)) and isinstance(expected, (float, int, Decimal)): return math.isclose(float(actual), float(expected), rel_tol=0, abs_tol=5.1e-7) return actual == expected def v1_price_position(close: float, base_low: float, base_high: float, optimistic_high: float) -> str: values = (close, base_low, base_high, optimistic_high) if not all(math.isfinite(value) for value in values): raise LedgerError("E_RANGE", "价格和区间必须是有限数字") if close <= 0 or base_low <= 0 or base_low > base_high or base_high > optimistic_high: raise LedgerError("E_RANGE", "价格必须为正且 base_low <= base_high <= optimistic_high") if close > optimistic_high: return "ABOVE_OPTIMISTIC_RANGE" if close > base_high: return "ABOVE_BASE_RANGE" if close >= base_low: return "WITHIN_BASE_RANGE" return "BELOW_BASE_RANGE" def judge_price(close: float, base_low: float, base_high: float, optimistic_high: float) -> str: return V1_POSITION_LABELS[v1_price_position(close, base_low, base_high, optimistic_high)] def load_price_fixture(path: Path) -> list[PriceRecord]: payload = _load_json(path) if payload.get("schema_version") != 1 or not isinstance(payload.get("prices"), list): raise LedgerError("E_INPUT_SCHEMA", "价格 fixture 必须是 schema_version=1 且包含 prices 数组") records: list[PriceRecord] = [] seen: set[str] = set() for index, item in enumerate(payload["prices"]): if not isinstance(item, dict): raise LedgerError("E_INPUT_SCHEMA", f"prices[{index}] 必须是对象") ticker = str(item.get("ticker") or "").upper() if not ticker or ticker in seen: raise LedgerError("E_INPUT_SCHEMA", f"prices[{index}].ticker 缺失或重复") seen.add(ticker) complete = item.get("is_complete_trading_day") is True records.append( PriceRecord( ticker=ticker, trade_date=_iso_date(item.get("trade_date"), f"prices[{index}].trade_date"), close=float(_finite_float(item.get("close"), f"prices[{index}].close")), currency=str(item.get("currency") or ""), source_id=str(item.get("source_id") or ""), source_timestamp=_canonical_timestamp(item.get("source_timestamp"), f"prices[{index}].source_timestamp"), is_complete_trading_day=complete, ) ) return records def _serialise_row(row: dict[str, Any]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in row.items(): if isinstance(value, Decimal): result[key] = float(value) elif isinstance(value, datetime): result[key] = value.isoformat() elif isinstance(value, date): result[key] = value.isoformat() else: result[key] = value return result def active_securities(settings: MySQLSettings) -> list[dict[str, str]]: connection = connect_database(settings) cursor = connection.cursor(dictionary=True) try: cursor.execute( "SELECT ticker, company, market, currency FROM security " "WHERE active=1 ORDER BY ticker" ) return [_serialise_row(row) for row in cursor.fetchall()] finally: cursor.close() connection.close() def apply_daily_prices( settings: MySQLSettings, as_of: str, prices: Iterable[PriceRecord], initial_failures: Sequence[dict[str, str]] = (), ) -> dict[str, Any]: cutoff = date.fromisoformat(_iso_date(as_of, "as_of")) connection = connect_database(settings) cursor = connection.cursor(dictionary=True) failures = [dict(item) for item in initial_failures] inserted_prices = 0 inserted_judgements = 0 unchanged = 0 now = datetime.now(timezone.utc).replace(tzinfo=None) try: connection.start_transaction() cursor.execute("SELECT ticker, currency FROM security WHERE active=1") securities = {row["ticker"]: row for row in cursor.fetchall()} for price in prices: try: security = securities.get(price.ticker) if security is None: raise LedgerError("E_UNKNOWN_TICKER", "证券未登记或未启用") if not price.is_complete_trading_day: raise LedgerError("E_INCOMPLETE_TRADING_DAY", "不是完整交易日收盘价") if not math.isfinite(price.close) or price.close <= 0: raise LedgerError("E_PRICE", "收盘价必须是有限正数") trade_date = date.fromisoformat(price.trade_date) if trade_date > cutoff: raise LedgerError("E_ASOF", "行情日期晚于请求 as-of") if price.currency != security["currency"]: raise LedgerError("E_CURRENCY", "行情币种与证券币种不一致") if not price.source_id: raise LedgerError("E_SOURCE", "行情 source_id 缺失") source_timestamp = _canonical_timestamp(price.source_timestamp, "source_timestamp") source_instant = datetime.fromisoformat(source_timestamp).astimezone(timezone.utc) cursor.execute( """ SELECT * FROM valuation_version WHERE ticker=%s AND active_from<=%s AND (active_to IS NULL OR active_to>=%s) ORDER BY valuation_date DESC """, (price.ticker, price.trade_date, price.trade_date), ) versions = cursor.fetchall() if len(versions) != 1: raise LedgerError("E_VALUATION_VERSION", f"有效估值版本数量必须为 1，实际 {len(versions)}") version = versions[0] base_low = float(version["base_low"]) base_high = float(version["base_high"]) optimistic_high = float(version["optimistic_high"]) label = judge_price(price.close, base_low, base_high, optimistic_high) expected = ( version["valuation_id"], price.close, base_low, base_high, optimistic_high, label, ) cursor.execute( "SELECT close, currency, source_id, source_timestamp FROM daily_price " "WHERE ticker=%s AND trade_date=%s", (price.ticker, price.trade_date), ) existing_price = cursor.fetchone() if existing_price: timestamp_db = source_instant.replace(tzinfo=None) if ( not math.isclose(float(existing_price["close"]), price.close, rel_tol=0, abs_tol=5.1e-7) or existing_price["currency"] != price.currency or existing_price["source_id"] != price.source_id or existing_price["source_timestamp"] != timestamp_db ): raise LedgerError("E_PRICE_IMMUTABLE", "同交易日历史收盘价冲突，拒绝改写") else: cursor.execute( """ INSERT INTO daily_price(ticker, trade_date, close, currency, source_id, source_timestamp, ingested_at) VALUES (%s, %s, %s, %s, %s, %s, %s) """, ( price.ticker, price.trade_date, price.close, price.currency, price.source_id, source_instant.replace(tzinfo=None), now, ), ) inserted_prices += 1 cursor.execute( """ SELECT valuation_id, close, base_low, base_high, optimistic_high, label FROM daily_judgement WHERE ticker=%s AND trade_date=%s """, (price.ticker, price.trade_date), ) existing_judgement = cursor.fetchone() if existing_judgement: actual = ( existing_judgement["valuation_id"], float(existing_judgement["close"]), float(existing_judgement["base_low"]), float(existing_judgement["base_high"]), float(existing_judgement["optimistic_high"]), existing_judgement["label"], ) if actual[0] != expected[0] or actual[5] != expected[5] or any( not math.isclose(actual[index], expected[index], rel_tol=0, abs_tol=5.1e-7) for index in range(1, 5) ): raise LedgerError("E_JUDGEMENT_IMMUTABLE", "历史判定与当前计算不一致，拒绝改写") unchanged += 1 continue cursor.execute( """ INSERT INTO daily_judgement( ticker, trade_date, valuation_id, close, base_low, base_high, optimistic_high, label, distance_to_base_low, distance_to_base_high, computed_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( price.ticker, price.trade_date, version["valuation_id"], price.close, base_low, base_high, optimistic_high, label, (price.close - base_low) / base_low, (price.close - base_high) / base_high, now, ), ) inserted_judgements += 1 except LedgerError as exc: failures.append({"ticker": price.ticker, "code": exc.code, "message": str(exc)}) connection.commit() except mysql.connector.Error as exc: connection.rollback() raise LedgerError("E_DATABASE", safe_mysql_error(exc)) from exc except BaseException: connection.rollback() raise finally: cursor.close() connection.close() if failures and inserted_judgements == 0: status = "FAILED" elif failures: status = "COMPLETE_WITH_GAPS" elif inserted_prices == 0 and inserted_judgements == 0: status = "NO_NEW_TRADING_DAY" else: status = "COMPLETE" return { "status": status, "as_of": cutoff.isoformat(), "inserted_prices": inserted_prices, "inserted_judgements": inserted_judgements, "unchanged": unchanged, "failure_count": len(failures), "failures": failures, } LATEST_QUERY = """ WITH chosen AS ( SELECT ticker, MAX(trade_date) AS trade_date FROM daily_judgement GROUP BY ticker ) SELECT d.ticker, s.company, s.market, s.currency, d.close, d.trade_date, d.base_low, d.base_high, d.optimistic_high, d.label, d.distance_to_base_low, d.distance_to_base_high, v.valuation_date, v.consensus_year, v.consensus_profit, v.consensus_count, v.report_path, v.snapshot_path, v.source_hash FROM chosen c JOIN daily_judgement d ON d.ticker=c.ticker AND d.trade_date=c.trade_date JOIN security s ON s.ticker=d.ticker JOIN valuation_version v ON v.valuation_id=d.valuation_id """ def list_rows( settings: MySQLSettings, selected_date: str = "latest", *, label: str | None = None, ticker: str | None = None, company: str | None = None, ) -> list[dict[str, Any]]: connection = connect_database(settings) cursor = connection.cursor(dictionary=True) try: if selected_date == "latest": query = LATEST_QUERY params: list[Any] = [] else: selected_date = _iso_date(selected_date, "date") query = LATEST_QUERY.replace( "SELECT ticker, MAX(trade_date) AS trade_date\n FROM daily_judgement\n GROUP BY ticker", "SELECT ticker, trade_date FROM daily_judgement WHERE trade_date=%s", ) params = [selected_date] filters: list[str] = [] if label: filters.append("d.label=%s") params.append(label) if ticker: filters.append("d.ticker=%s") params.append(ticker.upper()) if company: filters.append("s.company LIKE %s") params.append(f"%{company}%") if filters: query += " WHERE " + " AND ".join(filters) query += " ORDER BY d.ticker" cursor.execute(query, params) return [_serialise_row(row) for row in cursor.fetchall()] finally: cursor.close() connection.close() def show_ticker(settings: MySQLSettings, ticker: str) -> dict[str, Any]: ticker = ticker.upper() connection = connect_database(settings) cursor = connection.cursor(dictionary=True) try: cursor.execute("SELECT * FROM security WHERE ticker=%s", (ticker,)) security = cursor.fetchone() if security is None: raise LedgerError("E_UNKNOWN_TICKER", f"证券不存在：{ticker}") cursor.execute("SELECT * FROM valuation_version WHERE ticker=%s ORDER BY valuation_date", (ticker,)) versions = [_serialise_row(row) for row in cursor.fetchall()] cursor.execute("SELECT * FROM daily_price WHERE ticker=%s ORDER BY trade_date", (ticker,)) prices = [_serialise_row(row) for row in cursor.fetchall()] cursor.execute("SELECT * FROM daily_judgement WHERE ticker=%s ORDER BY trade_date", (ticker,)) judgements = [_serialise_row(row) for row in cursor.fetchall()] return { "security": _serialise_row(security), "valuation_versions": versions, "daily_prices": prices, "daily_judgements": judgements, } finally: cursor.close() connection.close() EXPORT_FIELDS = [ "ticker", "company", "market", "currency", "close", "trade_date", "base_low", "base_high", "optimistic_high", "label", "distance_to_base_low", "distance_to_base_high", "valuation_date", "consensus_year", "consensus_profit", "consensus_count", "report_path", ] def _write_pair(files: Sequence[tuple[Path, bytes]]) -> list[str]: if len(files) != 2 or len({path for path, _ in files}) != 2: raise LedgerError("E_EXPORT", "latest 导出必须是两个不同文件") stages: dict[Path, Path] = {} backups: dict[Path, Path] = {} placed: set[Path] = set() rollback_errors: list[BaseException] = [] try: for target, data in files: target.parent.mkdir(parents=True, exist_ok=True) if target.exists() and not target.is_file(): raise LedgerError("E_EXPORT_PATH", f"导出目标不是普通文件：{target}") descriptor, raw_stage = tempfile.mkstemp(prefix=".pair-stage-", dir=target.parent) stage = Path(raw_stage) stages[target] = stage try: handle = os.fdopen(descriptor, "wb") except BaseException as primary: try: os.close(descriptor) except BaseException as secondary: if hasattr(primary, "add_note"): primary.add_note(f"latest staging fd 回收异常：{type(secondary).__name__}") raise with handle: handle.write(data) handle.flush() os.fsync(handle.fileno()) for target, _ in files: if target.exists(): descriptor, raw_backup = tempfile.mkstemp(prefix=".pair-backup-", dir=target.parent) os.close(descriptor) backup = Path(raw_backup) backup.unlink() os.replace(target, backup) backups[target] = backup for target, _ in files: os.replace(stages[target], target) placed.add(target) except BaseException as primary: for target, _ in reversed(files): try: if target in placed and target.exists(): target.unlink() backup = backups.get(target) if backup is not None and backup.exists(): os.replace(backup, target) except BaseException as secondary: rollback_errors.append(secondary) for stage in stages.values(): try: if stage.exists(): stage.unlink() except BaseException as secondary: rollback_errors.append(secondary) for secondary in rollback_errors: if hasattr(primary, "add_note"): primary.add_note(f"latest 双文件回滚异常：{type(secondary).__name__}") raise warnings: list[str] = [] for backup in backups.values(): try: if backup.exists(): backup.unlink() except OSError: warnings.append("E_EXPORT_BACKUP_RETAINED") for stage in stages.values(): if stage.exists(): stage.unlink() return warnings def export_latest(settings: MySQLSettings, output_dir: Path, selected_date: str = "latest") -> dict[str, Any]: rows = list_rows(settings, selected_date) output_dir.mkdir(parents=True, exist_ok=True) csv_buffer = io.StringIO(newline="") writer = csv.DictWriter(csv_buffer, fieldnames=EXPORT_FIELDS, extrasaction="ignore", lineterminator="\n") writer.writeheader() writer.writerows(rows) csv_path = output_dir / "latest.csv" lines = [ "# 股票估值每日台账最新总表", "", f"- 生成时间：`{utc_now()}`", f"- 记录数：`{len(rows)}`", "- 口径：最近完整交易日收盘价相对最近有效正式估值版本；不构成交易指令。", "", "| 代码 | 公司 | 市场 | 币种 | 收盘价 | 交易日 | 基准区间 | 乐观上沿 | 判定 | 估值日 | 距基准下沿 | 距基准上沿 | 机构覆盖 | 正式报告 |", "|---|---|---|---|---:|---|---:|---:|---|---|---:|---:|---:|---|", ] for row in rows: coverage = "-" if row["consensus_count"] is None else str(row["consensus_count"]) lines.append( "| {ticker} | {company} | {market} | {currency} | {close:.4f} | {trade_date} | " "{base_low:.4f}—{base_high:.4f} | {optimistic_high:.4f} | {label} | {valuation_date} | " "{distance_to_base_low:.2%} | {distance_to_base_high:.2%} | {coverage} | `{report_path}` |".format( coverage=coverage, **row ) ) markdown_path = output_dir / "latest.md" warnings = _write_pair( ( (csv_path, csv_buffer.getvalue().encode("utf-8-sig")), (markdown_path, ("\n".join(lines) + "\n").encode("utf-8")), ) ) return { "status": "EXPORTED", "row_count": len(rows), "markdown": str(markdown_path), "csv": str(csv_path), "warnings": warnings, }