"""复权折算（v3 §20.5）测试：SQL 侧按 adjust_factor 折算价格列。 覆盖： - hfq：price × factor（后复权） - qfq：price × factor / 该股最新 factor（前复权，归一） - volume/amount 不折算 - 因子缺失按 1.0 兜底，并由 count_price_adjust_gaps 如实统计 - hfq 与 qfq 的**收益率序列完全一致**（仅差一个常数倍），这是复权口径自洽性的关键断言 """ from __future__ import annotations from datetime import date from decimal import Decimal import pytest from app.domain.entities.market import AdjustFactor, DailyBar from app.infrastructure.persistence.sqlalchemy.base import Base from app.infrastructure.persistence.sqlalchemy.repositories.market_impl import ( SqlAlchemyDailyBarRepository, ) from sqlalchemy import create_engine from sqlalchemy.orm import Session _D = [date(2024, 6, 3), date(2024, 6, 4), date(2024, 6, 5)] def _bar(day: date, close: str, symbol: str = "600519.SH") -> DailyBar: return DailyBar( symbol=symbol, trade_date=day, source="tushare", adjust="none", open=Decimal(close), high=Decimal(close), low=Decimal(close), close=Decimal(close), volume=Decimal("1000"), amount=Decimal("100000"), ) def _factor(day: date, factor: str, symbol: str = "600519.SH") -> AdjustFactor: return AdjustFactor(symbol=symbol, trade_date=day, factor=Decimal(factor), source="tushare") @pytest.fixture def session(tmp_path) -> Session: engine = create_engine(f"sqlite:///{tmp_path / 'adjprice.db'}", future=True) Base.metadata.create_all(engine) with Session(engine) as s: repo = SqlAlchemyDailyBarRepository(s) # 100 → 除权前 1.0；110 → 之后因子 1.1（模拟一次分红/送股） repo.upsert_many([_bar(_D[0], "100"), _bar(_D[1], "110"), _bar(_D[2], "121")]) s.add_all( [ _factor_model(_D[0], "1.0"), _factor_model(_D[1], "1.1"), _factor_model(_D[2], "1.1"), ] ) s.commit() yield s def _factor_model(day: date, factor: str, symbol: str = "600519.SH"): from app.infrastructure.persistence.sqlalchemy.models.market import AdjustFactorModel return AdjustFactorModel(symbol=symbol, trade_date=day, factor=Decimal(factor)) def _stream(session, price_adjust: str, columns=("close",)) -> dict[date, tuple]: repo = SqlAlchemyDailyBarRepository(session) rows = list( repo.stream_range_many_columns( ["600519.SH"], _D[0], _D[2], list(columns), adjust="none", price_adjust=price_adjust, ) ) return {date.fromisoformat(r[1]): r[2:] for r in rows} class TestAdjustedPrices: def test_hfq_multiplies_by_factor(self, session) -> None: out = _stream(session, "hfq") assert out[_D[0]] == (100.0,) # 100 × 1.0 assert out[_D[1]] == pytest.approx((121.0,)) # 110 × 1.1 assert out[_D[2]] == pytest.approx((133.1,)) # 121 × 1.1 def test_qfq_normalizes_by_latest_factor(self, session) -> None: out = _stream(session, "qfq") assert out[_D[0]] == pytest.approx((100 / 1.1,)) # 100 × 1.0 / 1.1 assert out[_D[1]] == pytest.approx((110.0,)) # 110 × 1.1 / 1.1 assert out[_D[2]] == pytest.approx((121.0,)) def test_volume_and_amount_not_adjusted(self, session) -> None: out = _stream(session, "hfq", columns=("close", "volume", "amount")) close, volume, amount = out[_D[2]] assert close == pytest.approx(133.1) assert volume == 1000.0 and amount == 100000.0 # 不随复权缩放 def test_none_is_raw(self, session) -> None: out = _stream(session, "none") assert out[_D[0]] == (100.0,) and out[_D[2]] == (121.0,) def test_hfq_and_qfq_give_identical_returns(self, session) -> None: hfq, qfq = _stream(session, "hfq"), _stream(session, "qfq") for a, b in zip(_D, _D[1:], strict=False): r_hfq = hfq[b][0] / hfq[a][0] r_qfq = qfq[b][0] / qfq[a][0] assert r_hfq == pytest.approx(r_qfq, rel=1e-12) def test_invalid_price_adjust_rejected(self, session) -> None: repo = SqlAlchemyDailyBarRepository(session) with pytest.raises(ValueError, match="复权"): list( repo.stream_range_many_columns( ["600519.SH"], _D[0], _D[2], ["close"], price_adjust="bad" ) ) class TestAdjustGapReporting: def test_missing_factor_falls_back_and_is_reported(self, session) -> None: repo = SqlAlchemyDailyBarRepository(session) # 追加一行无因子的行情（模拟 25 只蓝筹 2020-2022 缺因子的情形） repo.upsert_many([_bar(date(2024, 6, 6), "200")]) session.commit() out = dict( (date.fromisoformat(r[1]), r[2]) for r in repo.stream_range_many_columns( ["600519.SH"], _D[0], date(2024, 6, 6), ["close"], price_adjust="hfq" ) ) assert out[date(2024, 6, 6)] == 200.0 # 缺因子 → 系数 1.0（未折算） total, missing = repo.count_price_adjust_gaps( ["600519.SH"], _D[0], date(2024, 6, 6) ) assert total == 4 and missing == 1 def test_no_gap_when_all_covered(self, session) -> None: repo = SqlAlchemyDailyBarRepository(session) total, missing = repo.count_price_adjust_gaps(["600519.SH"], _D[0], _D[2]) assert total == 3 and missing == 0 def test_empty_symbols(self, session) -> None: repo = SqlAlchemyDailyBarRepository(session) assert repo.count_price_adjust_gaps([], _D[0], _D[2]) == (0, 0) class TestQfqMissingFactor: """qfq 缺因子行不得被错误缩放。 回归用例：`coalesce(factor,1)/latest` 会把缺口行缩放到 `1/latest` （latest=5 → 121/5=24.2，凭空 −80% 单日跌幅），正确写法必须把 COALESCE 放在最外层，使缺口行保持原始价（口径：缺失因子按 1.0 兜底、不折算）。 """ def test_qfq_gap_row_is_not_rescaled(self, session) -> None: from app.infrastructure.persistence.sqlalchemy.models.market import AdjustFactorModel session.query(AdjustFactorModel).delete() # 清掉 fixture 的 1.0/1.1/1.1 session.add_all([_factor_model(_D[0], "5.0"), _factor_model(_D[1], "5.0")]) session.commit() # 第 3 天无因子 → 缺口 out = _stream(session, "qfq") # 有因子的行：price × factor / max(factor=5.0) → 等于原始价 assert out[_D[0]] == pytest.approx((100.0,)) assert out[_D[1]] == pytest.approx((110.0,)) # 缺口行：保持原始价（若被错误归一则为 121/5 = 24.2） assert out[_D[2]] == pytest.approx((121.0,)) repo = SqlAlchemyDailyBarRepository(session) total, missing = repo.count_price_adjust_gaps(["600519.SH"], _D[0], _D[2]) assert (total, missing) == (3, 1)