66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
"""复权换算 adjust_bars：bfq/qfq/hfq 乘数、阶梯因子向前沿用、名义量不缩放。"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
|
||
import pytest
|
||
|
||
from app.api._deps import adjust_bars
|
||
from app.domain import Bar
|
||
|
||
|
||
def _bars(*rows) -> list[Bar]:
|
||
# (date, open, close)
|
||
return [Bar(ts=datetime(y, m, d), open=o, high=o, low=o, close=c, volume=100.0)
|
||
for (y, m, d), o, c in rows]
|
||
|
||
|
||
def _factors(*pairs):
|
||
return [(datetime(y, m, d), f) for (y, m, d), f in pairs]
|
||
|
||
|
||
def test_bfq_to_bfq_identity():
|
||
bars = _bars(((2024, 1, 2), 10.0, 11.0))
|
||
out = adjust_bars(bars, _factors(((2024, 1, 2), 2.0)), "bfq", "bfq")
|
||
assert out[0].close == 11.0
|
||
|
||
|
||
def test_qfq_normalizes_by_latest_factor():
|
||
# 因子 1.0 -> 2.0（中途除权）：qfq = f(t)/f(latest)
|
||
bars = _bars(
|
||
((2024, 1, 2), 10.0, 10.0), # f=1.0
|
||
((2024, 6, 3), 5.0, 5.0), # f=2.0（除权日，价格腰斩）
|
||
)
|
||
factors = _factors(((2024, 1, 2), 1.0), ((2024, 6, 3), 2.0))
|
||
out = adjust_bars(bars, factors, "bfq", "qfq")
|
||
# 除权前按 1.0/2.0 缩放 -> 5.0；除权后 2.0/2.0 -> 原价
|
||
assert out[0].close == pytest.approx(5.0)
|
||
assert out[1].close == pytest.approx(5.0)
|
||
|
||
|
||
def test_hfq_scales_by_factor():
|
||
bars = _bars(((2024, 1, 2), 10.0, 10.0), ((2024, 6, 3), 5.0, 5.0))
|
||
factors = _factors(((2024, 1, 2), 1.0), ((2024, 6, 3), 2.0))
|
||
out = adjust_bars(bars, factors, "bfq", "hfq")
|
||
assert out[0].close == pytest.approx(10.0) # f=1
|
||
assert out[1].close == pytest.approx(10.0) # 5 * 2
|
||
|
||
|
||
def test_factor_step_forward_fill():
|
||
"""因子是阶梯函数：变化点之间的日期向前沿用最近因子。"""
|
||
bars = _bars(((2024, 2, 1), 8.0, 8.0)) # 在 1/2 与 6/3 之间 -> 沿用 1.0
|
||
factors = _factors(((2024, 1, 2), 1.0), ((2024, 6, 3), 2.0))
|
||
out = adjust_bars(bars, factors, "bfq", "hfq")
|
||
assert out[0].close == pytest.approx(8.0) # 8 * 1.0
|
||
|
||
|
||
def test_nominal_columns_not_scaled():
|
||
"""成交额/换手率是名义量，不随复权缩放。"""
|
||
bars = [Bar(ts=datetime(2024, 1, 2), open=10, high=10, low=10, close=10,
|
||
volume=100.0, amount=1_000_000.0, turnover=1.5)]
|
||
out = adjust_bars(bars, _factors(((2024, 1, 2), 4.0)), "bfq", "hfq")
|
||
assert out[0].close == pytest.approx(40.0)
|
||
assert out[0].amount == 1_000_000.0
|
||
assert out[0].turnover == 1.5
|
||
assert out[0].volume == 100.0
|