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()