from __future__ import annotations
|
|
import contextlib
|
import hashlib
|
import io
|
import json
|
import os
|
import re
|
import sys
|
import tempfile
|
import time
|
import unittest
|
import uuid
|
from dataclasses import replace
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
from unittest import mock
|
|
import mysql.connector
|
|
|
ROOT = Path(__file__).resolve().parents[4]
|
PROJECT_DEV = ROOT / "dev" / "project-dev"
|
ANA_DEV = ROOT / "dev" / "ana-dev"
|
sys.path.insert(0, str(PROJECT_DEV))
|
sys.path.insert(0, str(ANA_DEV))
|
|
from stock_valuation_ledger import core # noqa: E402
|
from stock_valuation_ledger.cli import main # noqa: E402
|
from stock_valuation_ledger.core import ( # noqa: E402
|
LedgerError,
|
PriceRecord,
|
ValuationCandidate,
|
_write_pair,
|
active_securities,
|
apply_daily_prices,
|
export_latest,
|
import_valuations,
|
initialize_database,
|
judge_price,
|
list_rows,
|
show_ticker,
|
)
|
from stock_valuation_ledger.database import MySQLSettings # noqa: E402
|
from stock_valuation_ledger.market import ( # noqa: E402
|
_market_schema_contract,
|
_record_from_row,
|
)
|
|
|
RESULTS_ROOT = ROOT / "ana-data" / "result" / "股票估值"
|
SHANGHAI = timezone(timedelta(hours=8))
|
|
|
def _mysql_env() -> dict[str, str]:
|
names = {
|
"host": "STOCK_VALUATION_MYSQL_HOST",
|
"port": "STOCK_VALUATION_MYSQL_PORT",
|
"user": "STOCK_VALUATION_MYSQL_USER",
|
"password": "STOCK_VALUATION_MYSQL_PASSWORD",
|
}
|
values = {key: os.environ.get(name) for key, name in names.items()}
|
if any(value is None for value in values.values()):
|
raise unittest.SkipTest("MySQL integration environment is not configured")
|
return {key: str(value) for key, value in values.items()}
|
|
|
class LedgerMySQLTests(unittest.TestCase):
|
@classmethod
|
def setUpClass(cls) -> None:
|
cfg = _mysql_env()
|
cls.server = {
|
"host": cfg["host"],
|
"port": int(cfg["port"]),
|
"user": cfg["user"],
|
"password": cfg["password"],
|
}
|
cls.database = "stock_valuation_test_" + uuid.uuid4().hex[:12]
|
if not re.fullmatch(r"stock_valuation_test_[0-9a-f]{12}", cls.database):
|
raise AssertionError("unsafe test database name")
|
connection = mysql.connector.connect(**cls.server)
|
cursor = connection.cursor()
|
try:
|
cursor.execute(
|
f"CREATE DATABASE `{cls.database}` CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci"
|
)
|
finally:
|
cursor.close()
|
connection.close()
|
cls.settings = MySQLSettings(
|
cfg["host"], int(cfg["port"]), cfg["user"], cfg["password"],
|
cls.database, "trading_xuntou",
|
)
|
os.environ["STOCK_VALUATION_ALLOW_TEST_DATABASE"] = "1"
|
initialize_database(cls.settings)
|
|
@classmethod
|
def tearDownClass(cls) -> None:
|
name = getattr(cls, "database", "")
|
if not re.fullmatch(r"stock_valuation_test_[0-9a-f]{12}", name):
|
raise AssertionError("refusing unsafe database cleanup")
|
connection = mysql.connector.connect(**cls.server)
|
cursor = connection.cursor()
|
try:
|
cursor.execute(f"DROP DATABASE `{name}`")
|
finally:
|
cursor.close()
|
connection.close()
|
|
def setUp(self) -> None:
|
self.temp = tempfile.TemporaryDirectory()
|
self.root = Path(self.temp.name)
|
connection = self._connect()
|
cursor = connection.cursor()
|
try:
|
cursor.execute("SET FOREIGN_KEY_CHECKS=0")
|
for table in ("daily_judgement", "daily_price", "valuation_version", "security"):
|
cursor.execute(f"TRUNCATE TABLE `{table}`")
|
cursor.execute("SET FOREIGN_KEY_CHECKS=1")
|
connection.commit()
|
finally:
|
cursor.close()
|
connection.close()
|
|
def tearDown(self) -> None:
|
self.temp.cleanup()
|
|
def _connect(self):
|
return mysql.connector.connect(**self.server, database=self.database)
|
|
def _import_formal(self) -> dict:
|
return import_valuations(
|
self.settings, RESULTS_ROOT, ROOT, self.root / "import_gaps.csv"
|
)
|
|
def _candidate(
|
self,
|
*,
|
valuation_date: str = "2026-08-04",
|
source_hash: str = "a" * 64,
|
base_low: float = 10.0,
|
base_high: float = 20.0,
|
optimistic_high: float = 30.0,
|
) -> ValuationCandidate:
|
return ValuationCandidate(
|
ticker="300450.SZ", company="测试公司", market="深交所", currency="CNY",
|
valuation_date=valuation_date, method="PE",
|
pessimistic_low=5.0, pessimistic_high=9.0,
|
base_low=base_low, base_high=base_high,
|
optimistic_low=base_high, optimistic_high=optimistic_high,
|
normalized_profit=100.0, normalized_pe=20.0, pb=2.0, ps=3.0,
|
consensus_year=2026, consensus_profit=120.0, consensus_count=3,
|
report_path=f"reports/{valuation_date}.md",
|
snapshot_path=f"snapshots/{valuation_date}.json",
|
source_hash=source_hash,
|
)
|
|
def _import_candidates(self, rows: list[ValuationCandidate]) -> dict:
|
with mock.patch.object(core, "discover_valuations", return_value=(rows, [])):
|
return import_valuations(
|
self.settings, RESULTS_ROOT, ROOT, self.root / "import_gaps.csv"
|
)
|
|
def _cli(self, *args: str) -> tuple[int, dict]:
|
stream = io.StringIO()
|
with contextlib.redirect_stdout(stream):
|
code = main([*args, "--database", self.database])
|
return code, json.loads(stream.getvalue())
|
|
def test_01_schema_is_mysql_four_tables_and_constraints(self) -> None:
|
initialize_database(self.settings)
|
connection = self._connect()
|
cursor = connection.cursor(dictionary=True)
|
try:
|
cursor.execute("SHOW TABLE STATUS")
|
tables = {row["Name"]: row["Engine"] for row in cursor.fetchall()}
|
self.assertEqual(set(tables), set(core.EXPECTED_TABLE_COLUMNS))
|
self.assertEqual(set(tables.values()), {"InnoDB"})
|
cursor.execute(
|
"SELECT COUNT(*) AS n FROM information_schema.REFERENTIAL_CONSTRAINTS "
|
"WHERE CONSTRAINT_SCHEMA=%s",
|
(self.database,),
|
)
|
self.assertEqual(cursor.fetchone()["n"], 4)
|
finally:
|
cursor.close()
|
connection.close()
|
|
def test_02_formal_214_225_cold_warm_import(self) -> None:
|
cold = self._import_formal()
|
warm = self._import_formal()
|
self.assertEqual((cold["inserted"], cold["discovered"]), (225, 225))
|
self.assertEqual((warm["inserted"], warm["unchanged"]), (0, 224))
|
self.assertEqual(warm["gap_count"], 2)
|
connection = self._connect()
|
cursor = connection.cursor()
|
try:
|
cursor.execute("SELECT COUNT(*) FROM security")
|
self.assertEqual(cursor.fetchone()[0], 214)
|
cursor.execute("SELECT COUNT(*) FROM valuation_version")
|
self.assertEqual(cursor.fetchone()[0], 225)
|
batch = json.loads(
|
(RESULTS_ROOT / "20260805_batch_six_images_valuation" / "batch_results_83.json").read_text(encoding="utf-8")
|
)
|
batch_tickers = {str(row["ticker"]).upper() for row in batch["rows"]}
|
cursor.execute("SELECT ticker FROM security")
|
imported = {row[0] for row in cursor.fetchall()}
|
self.assertEqual(len(batch_tickers), 83)
|
self.assertTrue(batch_tickers <= imported)
|
finally:
|
cursor.close()
|
connection.close()
|
|
def test_03_all_securities_offline_daily_and_idempotency(self) -> None:
|
self._import_formal()
|
trade_date = "2026-08-06"
|
connection = self._connect()
|
cursor = connection.cursor(dictionary=True)
|
try:
|
cursor.execute(
|
"SELECT s.ticker,s.currency,v.base_low,v.base_high,v.optimistic_high "
|
"FROM security s JOIN valuation_version v ON v.ticker=s.ticker "
|
"AND v.active_from<=%s AND (v.active_to IS NULL OR v.active_to>=%s) "
|
"ORDER BY s.ticker",
|
(trade_date, trade_date),
|
)
|
rows = cursor.fetchall()
|
finally:
|
cursor.close()
|
connection.close()
|
self.assertEqual(len(rows), 214)
|
prices: list[PriceRecord] = []
|
expected_labels = set()
|
for index, row in enumerate(rows):
|
low, high, optimistic = map(float, (row["base_low"], row["base_high"], row["optimistic_high"]))
|
close = (low * 0.8, (low + high) / 2, (high + optimistic) / 2, optimistic * 1.2)[index % 4]
|
expected_labels.add(judge_price(close, low, high, optimistic))
|
prices.append(
|
PriceRecord(
|
row["ticker"], trade_date, close, row["currency"], "offline.fixture",
|
f"{trade_date}T16:00:00+08:00", True,
|
)
|
)
|
first = apply_daily_prices(self.settings, trade_date, prices)
|
second = apply_daily_prices(self.settings, trade_date, prices)
|
self.assertEqual((first["inserted_prices"], first["inserted_judgements"]), (214, 214))
|
self.assertEqual((second["inserted_prices"], second["unchanged"]), (0, 214))
|
self.assertEqual(expected_labels, {"偏低", "基本合理", "偏贵", "明显偏贵"})
|
|
def test_04_v1_judgement_labels_exact(self) -> None:
|
self.assertEqual(judge_price(9.99, 10, 20, 30), "偏低")
|
self.assertEqual(judge_price(10, 10, 20, 30), "基本合理")
|
self.assertEqual(judge_price(20, 10, 20, 30), "基本合理")
|
self.assertEqual(judge_price(20.01, 10, 20, 30), "偏贵")
|
self.assertEqual(judge_price(30, 10, 20, 30), "偏贵")
|
self.assertEqual(judge_price(30.01, 10, 20, 30), "明显偏贵")
|
|
def test_05_later_revaluation_effective_next_day(self) -> None:
|
old = self._candidate(valuation_date="2026-08-04", source_hash="1" * 64)
|
new = self._candidate(
|
valuation_date="2026-08-06", source_hash="2" * 64,
|
base_low=100.0, base_high=120.0, optimistic_high=150.0,
|
)
|
self._import_candidates([old, new])
|
connection = self._connect()
|
cursor = connection.cursor(dictionary=True)
|
try:
|
cursor.execute(
|
"SELECT valuation_date,active_from,active_to FROM valuation_version ORDER BY valuation_date"
|
)
|
rows = cursor.fetchall()
|
finally:
|
cursor.close()
|
connection.close()
|
self.assertEqual(str(rows[0]["active_from"]), "2026-08-04")
|
self.assertEqual(str(rows[0]["active_to"]), "2026-08-06")
|
self.assertEqual(str(rows[1]["active_from"]), "2026-08-07")
|
for day in ("2026-08-06", "2026-08-07"):
|
result = apply_daily_prices(
|
self.settings, day,
|
[PriceRecord("300450.SZ", day, 15, "CNY", "fixture", f"{day}T16:00:00+08:00")],
|
)
|
self.assertEqual(result["inserted_judgements"], 1)
|
rows = list_rows(self.settings, "2026-08-06") + list_rows(self.settings, "2026-08-07")
|
self.assertEqual([row["label"] for row in rows], ["基本合理", "偏低"])
|
|
def test_06_import_detects_results_change_under_same_snapshot_hash(self) -> None:
|
original = self._candidate()
|
self._import_candidates([original])
|
changed = replace(original, base_low=11.0)
|
result = self._import_candidates([changed])
|
self.assertEqual((result["inserted"], result["unchanged"], result["gap_count"]), (0, 0, 1))
|
gaps = (self.root / "import_gaps.csv").read_text(encoding="utf-8")
|
self.assertIn("E_IMMUTABLE_VERSION_CONFLICT", gaps)
|
connection = self._connect()
|
cursor = connection.cursor()
|
try:
|
cursor.execute("SELECT base_low FROM valuation_version")
|
self.assertEqual(float(cursor.fetchone()[0]), 10.0)
|
finally:
|
cursor.close()
|
connection.close()
|
|
def test_07_missing_version_has_no_price_half_state(self) -> None:
|
self._import_candidates([self._candidate(valuation_date="2026-08-06")])
|
result = apply_daily_prices(
|
self.settings, "2026-08-05",
|
[PriceRecord("300450.SZ", "2026-08-05", 15, "CNY", "fixture", "2026-08-05T16:00:00+08:00")],
|
)
|
self.assertEqual(result["failures"][0]["code"], "E_VALUATION_VERSION")
|
connection = self._connect()
|
cursor = connection.cursor()
|
try:
|
cursor.execute("SELECT COUNT(*) FROM daily_price")
|
self.assertEqual(cursor.fetchone()[0], 0)
|
finally:
|
cursor.close()
|
connection.close()
|
|
def test_08_same_day_price_is_immutable(self) -> None:
|
self._import_candidates([self._candidate()])
|
first = PriceRecord("300450.SZ", "2026-08-05", 15, "CNY", "fixture", "2026-08-05T16:00:00+08:00")
|
apply_daily_prices(self.settings, "2026-08-05", [first])
|
result = apply_daily_prices(self.settings, "2026-08-05", [replace(first, close=16)])
|
self.assertEqual(result["failures"][0]["code"], "E_PRICE_IMMUTABLE")
|
|
def test_09_list_show_export_consistency(self) -> None:
|
self._import_candidates([self._candidate()])
|
apply_daily_prices(
|
self.settings, "2026-08-05",
|
[PriceRecord("300450.SZ", "2026-08-05", 15, "CNY", "fixture", "2026-08-05T16:00:00+08:00")],
|
)
|
rows = list_rows(self.settings)
|
shown = show_ticker(self.settings, "300450.SZ")
|
exported = export_latest(self.settings, self.root)
|
self.assertEqual(rows[0]["ticker"], shown["security"]["ticker"])
|
self.assertEqual(exported["row_count"], 1)
|
self.assertIn("300450.SZ", (self.root / "latest.md").read_text(encoding="utf-8"))
|
|
def test_10_front_adjusted_row_contract_requires_complete_evidence(self) -> None:
|
row = {
|
"symbol": "300450.SZ",
|
"trade_date": "2026-08-04", "close": "45.67",
|
"source": "xtquant", "updated_at": "2026-08-04T16:10:00",
|
"is_open": 1, "calendar_source_kind": "xtdata.get_trading_dates",
|
"window_end": "2026-08-10", "asset_status": "current", "asset_is_current": 1,
|
}
|
security = {"ticker": "300450.SZ", "currency": "CNY"}
|
record = _record_from_row(
|
row, security, datetime(2026, 8, 5, tzinfo=SHANGHAI).date(),
|
datetime(2026, 8, 5, 9, tzinfo=SHANGHAI),
|
)
|
self.assertEqual((record.trade_date, record.close), ("2026-08-04", 45.67))
|
self.assertEqual(record.source_id, "trading_xuntou.cn_stock_kline_1d_front:xtquant:front")
|
bj_row = {**row, "symbol": "920185.BJ"}
|
bj_record = _record_from_row(
|
bj_row, {"ticker": "920185.BJ", "currency": "CNY"},
|
datetime(2026, 8, 5, tzinfo=SHANGHAI).date(),
|
datetime(2026, 8, 5, 9, tzinfo=SHANGHAI),
|
)
|
self.assertEqual(bj_record.ticker, "920185.BJ")
|
cases = (
|
({"updated_at": "2026-08-04T14:59:00"}, "E_INCOMPLETE_TRADING_DAY"),
|
({"close": 0}, "E_PRICE"),
|
({"symbol": "300450.SH"}, "E_MARKET_IDENTITY"),
|
({"symbol": "300450.sz"}, "E_MARKET_IDENTITY"),
|
({"source": "other"}, "E_SOURCE"),
|
({"source": "XTQUANT"}, "E_SOURCE"),
|
({"asset_is_current": 0}, "E_INCOMPLETE_TRADING_DAY"),
|
({"window_end": "2026-08-04"}, "E_INCOMPLETE_TRADING_DAY"),
|
)
|
for changes, code in cases:
|
with self.subTest(changes=changes):
|
with self.assertRaises(LedgerError) as captured:
|
_record_from_row(
|
{**row, **changes}, security, datetime(2026, 8, 5, tzinfo=SHANGHAI).date(),
|
datetime(2026, 8, 5, 9, tzinfo=SHANGHAI),
|
)
|
self.assertEqual(captured.exception.code, code)
|
|
def test_11_market_schema_failure_precedes_target_write_and_export(self) -> None:
|
self._import_candidates([self._candidate()])
|
output = self.root / "out"
|
with mock.patch(
|
"stock_valuation_ledger.market.fetch_trading_closes",
|
side_effect=LedgerError("E_MARKET_SCHEMA", "missing proof"),
|
):
|
code, result = self._cli("daily", "--as-of", "2026-08-05", "--output-dir", str(output))
|
self.assertEqual((code, result["error_code"]), (2, "E_MARKET_SCHEMA"))
|
self.assertFalse(output.exists())
|
connection = self._connect()
|
cursor = connection.cursor()
|
try:
|
cursor.execute("SELECT COUNT(*) FROM daily_price")
|
self.assertEqual(cursor.fetchone()[0], 0)
|
finally:
|
cursor.close()
|
connection.close()
|
|
def test_12_row_level_market_gap_preserves_good_row(self) -> None:
|
first = self._candidate(source_hash="1" * 64)
|
second = replace(first, ticker="600000.SH", company="第二公司", market="上交所", source_hash="2" * 64)
|
self._import_candidates([first, second])
|
good = PriceRecord("300450.SZ", "2026-08-05", 15, "CNY", "trading_xuntou.test:none", "2026-08-05T16:10:00+08:00")
|
market_result = ([good], [{"ticker": "600000.SH", "code": "E_INCOMPLETE_TRADING_DAY", "message": "incomplete"}], {"provider": "test"})
|
with mock.patch("stock_valuation_ledger.market.fetch_trading_closes", return_value=market_result):
|
code, result = self._cli("daily", "--as-of", "2026-08-05", "--output-dir", str(self.root / "out"))
|
self.assertEqual(code, 4)
|
self.assertEqual((result["inserted_prices"], result["inserted_judgements"]), (1, 1))
|
|
def test_13_all_failed_daily_preserves_latest_pair(self) -> None:
|
self._import_candidates([self._candidate()])
|
fixture = self.root / "empty.json"
|
fixture.write_text('{"schema_version":1,"prices":[]}\n', encoding="utf-8")
|
output = self.root / "out"
|
code, result = self._cli("daily", "--as-of", "2026-08-05", "--prices", str(fixture), "--output-dir", str(output))
|
self.assertEqual((code, result["status"]), (4, "FAILED"))
|
self.assertFalse(output.exists())
|
good = self.root / "good.json"
|
good.write_text(json.dumps({"schema_version": 1, "prices": [{
|
"ticker": "300450.SZ", "trade_date": "2026-08-05", "close": 15,
|
"currency": "CNY", "source_id": "fixture", "source_timestamp": "2026-08-05T16:00:00+08:00",
|
"is_complete_trading_day": True,
|
}]}), encoding="utf-8")
|
self._cli("daily", "--as-of", "2026-08-05", "--prices", str(good), "--output-dir", str(output))
|
before = {name: (output / name).read_bytes() for name in ("latest.csv", "latest.md")}
|
self._cli("daily", "--as-of", "2026-08-06", "--prices", str(fixture), "--output-dir", str(output))
|
self.assertEqual(before, {name: (output / name).read_bytes() for name in before})
|
|
def test_14_latest_pair_rolls_back_first_second_and_baseexceptions(self) -> None:
|
targets = ((self.root / "latest.csv", b"new-csv"), (self.root / "latest.md", b"new-md"))
|
for path, old in ((targets[0][0], b"old-csv"), (targets[1][0], b"old-md")):
|
path.write_bytes(old)
|
real_replace = os.replace
|
for failed_target in ("latest.csv", "latest.md"):
|
for exc in (OSError("fault"), KeyboardInterrupt(), SystemExit(9)):
|
with self.subTest(target=failed_target, exception=type(exc).__name__):
|
targets[0][0].write_bytes(b"old-csv")
|
targets[1][0].write_bytes(b"old-md")
|
def injected(src, dst, *, _target=failed_target, _exc=exc):
|
if Path(src).name.startswith(".pair-stage-") and Path(dst).name == _target:
|
raise _exc
|
return real_replace(src, dst)
|
with mock.patch.object(core.os, "replace", side_effect=injected):
|
with self.assertRaises(type(exc)):
|
_write_pair(targets)
|
self.assertEqual(targets[0][0].read_bytes(), b"old-csv")
|
self.assertEqual(targets[1][0].read_bytes(), b"old-md")
|
self.assertEqual(list(self.root.glob(".pair-stage-*")), [])
|
self.assertEqual(list(self.root.glob(".pair-backup-*")), [])
|
|
def test_15_market_module_is_select_only_and_has_no_v2_network_path(self) -> None:
|
source = (PROJECT_DEV / "stock_valuation_ledger" / "market.py").read_text(encoding="utf-8")
|
self.assertNotIn("stock_valuation_pipeline_v2", source)
|
self.assertNotRegex(source.upper(), r"\b(INSERT|UPDATE|DELETE|CREATE|DROP|ALTER)\b")
|
self.assertIn("cn_stock_kline_1d_front", source)
|
self.assertNotRegex(source, r"cn_stock_kline_1d(?!_front)")
|
self.assertNotIn("dividend_type", source)
|
self.assertNotIn("cn_stock_instrument_static", source)
|
self.assertIn("MAX(c.trade_date)", source)
|
self.assertIn("r.trade_date=e.trade_date", source)
|
self.assertIn("connect_market_readonly", source)
|
|
def test_16_credentials_are_not_literal_and_sqlite_backend_is_absent(self) -> None:
|
paths = list((PROJECT_DEV / "stock_valuation_ledger").glob("*"))
|
text = "\n".join(
|
path.read_text(encoding="utf-8") for path in paths
|
if path.is_file() and path.suffix in {".py", ".sql", ".md", ".ps1"}
|
)
|
self.assertNotRegex(text, r"(?i)password\s*=\s*['\"][^'\"]+['\"]")
|
self.assertNotRegex(text, r"mysql(?:\+\w+)?://[^\s:]+:[^\s@]+@")
|
self.assertNotRegex(text, r"(?i)sqlite|stock_valuation\.sqlite3|--db\b")
|
|
def test_17_warm_200_row_query_under_one_second(self) -> None:
|
candidate = self._candidate()
|
rows = []
|
for index in range(200):
|
ticker = f"{index:06d}.SZ"
|
rows.append(replace(candidate, ticker=ticker, company=f"公司{index}", source_hash=f"{index:064x}"))
|
self._import_candidates(rows)
|
prices = [
|
PriceRecord(row.ticker, "2026-08-05", 15, "CNY", "fixture", "2026-08-05T16:00:00+08:00")
|
for row in rows
|
]
|
apply_daily_prices(self.settings, "2026-08-05", prices)
|
list_rows(self.settings)
|
start = time.perf_counter()
|
result = list_rows(self.settings)
|
elapsed = time.perf_counter() - start
|
self.assertEqual(len(result), 200)
|
self.assertLess(elapsed, 1.0)
|
|
def test_18_mysql_batch_failure_rolls_back_prices_and_judgements(self) -> None:
|
first = self._candidate(source_hash="1" * 64)
|
second = replace(
|
first, ticker="600000.SH", company="第二公司", market="上交所",
|
source_hash="2" * 64,
|
)
|
self._import_candidates([first, second])
|
connection = self._connect()
|
cursor = connection.cursor()
|
try:
|
cursor.execute(
|
"CREATE TRIGGER test_fail_judgement BEFORE INSERT ON daily_judgement "
|
"FOR EACH ROW BEGIN IF NEW.ticker='600000.SH' THEN "
|
"SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='isolated test fault'; END IF; END"
|
)
|
connection.commit()
|
finally:
|
cursor.close()
|
connection.close()
|
prices = [
|
PriceRecord(ticker, "2026-08-05", 15, "CNY", "fixture", "2026-08-05T16:00:00+08:00")
|
for ticker in ("300450.SZ", "600000.SH")
|
]
|
with self.assertRaises(LedgerError) as captured:
|
apply_daily_prices(self.settings, "2026-08-05", prices)
|
self.assertEqual(captured.exception.code, "E_DATABASE")
|
connection = self._connect()
|
cursor = connection.cursor()
|
try:
|
cursor.execute("SELECT COUNT(*) FROM daily_price")
|
self.assertEqual(cursor.fetchone()[0], 0)
|
cursor.execute("SELECT COUNT(*) FROM daily_judgement")
|
self.assertEqual(cursor.fetchone()[0], 0)
|
finally:
|
cursor.close()
|
connection.close()
|
|
def test_19_stage_create_write_flush_fsync_faults_leave_no_pair_debris(self) -> None:
|
targets = ((self.root / "latest.csv", b"new-csv"), (self.root / "latest.md", b"new-md"))
|
real_mkstemp = tempfile.mkstemp
|
real_fdopen = os.fdopen
|
real_fsync = os.fsync
|
|
class FaultingHandle:
|
def __init__(self, handle, point, exc):
|
self.handle = handle
|
self.point = point
|
self.exc = exc
|
|
def __enter__(self):
|
self.handle.__enter__()
|
return self
|
|
def __exit__(self, *args):
|
return self.handle.__exit__(*args)
|
|
def write(self, data):
|
if self.point == "write":
|
raise self.exc
|
return self.handle.write(data)
|
|
def flush(self):
|
if self.point == "flush":
|
raise self.exc
|
return self.handle.flush()
|
|
def fileno(self):
|
return self.handle.fileno()
|
|
exception_factories = (
|
(OSError, lambda: OSError("fault")),
|
(KeyboardInterrupt, KeyboardInterrupt),
|
(SystemExit, lambda: SystemExit(9)),
|
)
|
for stage_number in (1, 2):
|
for point in ("create", "write", "flush", "fsync"):
|
for exc_type, factory in exception_factories:
|
with self.subTest(stage=stage_number, point=point, exception=exc_type.__name__):
|
targets[0][0].write_bytes(b"old-csv")
|
targets[1][0].write_bytes(b"old-md")
|
exc = factory()
|
stage_create_count = 0
|
stage_open_count = 0
|
stage_fsync_count = 0
|
|
def injected_mkstemp(*args, **kwargs):
|
nonlocal stage_create_count
|
if kwargs.get("prefix") == ".pair-stage-":
|
stage_create_count += 1
|
if point == "create" and stage_create_count == stage_number:
|
raise exc
|
return real_mkstemp(*args, **kwargs)
|
|
def injected_fdopen(descriptor, mode):
|
nonlocal stage_open_count
|
stage_open_count += 1
|
handle = real_fdopen(descriptor, mode)
|
fault_point = point if stage_open_count == stage_number and point in {"write", "flush"} else None
|
return FaultingHandle(handle, fault_point, exc)
|
|
def injected_fsync(descriptor):
|
nonlocal stage_fsync_count
|
stage_fsync_count += 1
|
if point == "fsync" and stage_fsync_count == stage_number:
|
raise exc
|
return real_fsync(descriptor)
|
|
with mock.patch.object(core.tempfile, "mkstemp", side_effect=injected_mkstemp), \
|
mock.patch.object(core.os, "fdopen", side_effect=injected_fdopen), \
|
mock.patch.object(core.os, "fsync", side_effect=injected_fsync):
|
with self.assertRaises(exc_type):
|
_write_pair(targets)
|
self.assertEqual(targets[0][0].read_bytes(), b"old-csv")
|
self.assertEqual(targets[1][0].read_bytes(), b"old-md")
|
self.assertEqual(list(self.root.glob(".pair-stage-*")), [])
|
self.assertEqual(list(self.root.glob(".pair-backup-*")), [])
|
|
def test_20_nonpositive_range_fails_before_price_insert(self) -> None:
|
self._import_candidates([self._candidate(base_low=0.0)])
|
result = apply_daily_prices(
|
self.settings,
|
"2026-08-05",
|
[PriceRecord("300450.SZ", "2026-08-05", 15, "CNY", "fixture", "2026-08-05T16:00:00+08:00")],
|
)
|
self.assertEqual((result["status"], result["inserted_prices"], result["inserted_judgements"]), ("FAILED", 0, 0))
|
self.assertEqual(result["failures"][0]["code"], "E_RANGE")
|
connection = self._connect()
|
cursor = connection.cursor()
|
try:
|
cursor.execute("SELECT COUNT(*) FROM daily_price")
|
self.assertEqual(cursor.fetchone()[0], 0)
|
cursor.execute("SELECT COUNT(*) FROM daily_judgement")
|
self.assertEqual(cursor.fetchone()[0], 0)
|
finally:
|
cursor.close()
|
connection.close()
|
|
|
class MarketSchemaUnitTests(unittest.TestCase):
|
def test_21_front_table_missing_required_schema_proof_is_blocking(self) -> None:
|
class Cursor:
|
def execute(self, *_args, **_kwargs):
|
return None
|
def fetchall(self):
|
rows = []
|
for table, columns in core.EXPECTED_TABLE_COLUMNS.items():
|
del table, columns
|
for table, columns in {
|
"cn_stock_kline_1d_front": {"id", "symbol", "trade_date", "close", "updated_at"},
|
"formal_trading_calendar_days": {"market", "trade_date", "is_open", "source_kind", "source_fetch_time", "source_version", "asset_version"},
|
"formal_trading_calendar_assets": {"asset_version", "market", "window_end", "generated_at", "status", "is_current"},
|
}.items():
|
rows.extend({"TABLE_NAME": table, "COLUMN_NAME": column, "IS_NULLABLE": "NO", "COLUMN_DEFAULT": None, "COLUMN_COMMENT": ""} for column in columns)
|
return rows
|
def close(self):
|
return None
|
class Connection:
|
def cursor(self, dictionary=False):
|
return Cursor()
|
with self.assertRaises(LedgerError) as captured:
|
_market_schema_contract(Connection(), "trading_xuntou")
|
self.assertEqual(captured.exception.code, "E_MARKET_SCHEMA")
|
|
def test_22_front_table_schema_contract_is_explicit(self) -> None:
|
class Cursor:
|
def execute(self, *_args, **_kwargs):
|
return None
|
def fetchall(self):
|
rows = []
|
for table, columns in {
|
"cn_stock_kline_1d_front": {"id", "symbol", "trade_date", "close", "source", "updated_at"},
|
"formal_trading_calendar_days": {"market", "trade_date", "is_open", "source_kind", "source_fetch_time", "source_version", "asset_version"},
|
"formal_trading_calendar_assets": {"asset_version", "market", "window_end", "generated_at", "status", "is_current"},
|
}.items():
|
rows.extend({"TABLE_NAME": table, "COLUMN_NAME": column, "IS_NULLABLE": "NO", "COLUMN_DEFAULT": None, "COLUMN_COMMENT": ""} for column in columns)
|
return rows
|
def close(self):
|
return None
|
class Connection:
|
def cursor(self, dictionary=False):
|
return Cursor()
|
contract = _market_schema_contract(Connection(), "trading_xuntou")
|
self.assertEqual(contract["price_table"], "cn_stock_kline_1d_front")
|
self.assertEqual(contract["adjustment_semantics"], "front_adjusted_dedicated_table")
|
self.assertEqual(contract["source_id"], "trading_xuntou.cn_stock_kline_1d_front:xtquant:front")
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
