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,
|
}
|
