Yuxuan Yan
528620b271
- pyproject.toml: fail_under 80 → 95 - test_alpha: +79 lines - test_cli_workflow: +226 lines - test_derived: +121 lines - test_downloader_contracts: +169 lines - test_features: +16 lines - test_minute_downloader: +81 lines - test_portfolio: +208 lines
584 lines
20 KiB
Python
584 lines
20 KiB
Python
"""Tests for pipeline alpha computation and combination (no network)."""
|
|
import textwrap
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from pipeline.alpha.base import BaseAlpha
|
|
from pipeline.alpha.compute import (
|
|
compute_alpha,
|
|
evaluate_alpha,
|
|
investable_universe_mask,
|
|
)
|
|
from pipeline.alpha.registry import (
|
|
available_alphas,
|
|
get_alpha,
|
|
load_alpha_module,
|
|
register_alpha,
|
|
)
|
|
from pipeline.combo.combine import combine_alphas, _equal_weight
|
|
from pipeline.common.schema import ALPHA_COLUMNS, COMBO_COLUMNS
|
|
|
|
|
|
def _make_data(n_days: int = 30, symbols=("sh600000", "sz000001", "sh600519")) -> pd.DataFrame:
|
|
"""Build a synthetic long-format DATA_COLUMNS frame with deterministic prices."""
|
|
dates = pd.date_range("2024-01-01", periods=n_days)
|
|
rng = np.random.default_rng(0)
|
|
frames = []
|
|
for i, sym in enumerate(symbols):
|
|
# Distinct drift per symbol so the cross-section is non-degenerate.
|
|
close = 100.0 + i * 5 + np.cumsum(rng.standard_normal(n_days))
|
|
frames.append(pd.DataFrame({
|
|
"symbol_id": sym,
|
|
"symbol_name": sym,
|
|
"date": dates,
|
|
"open": close,
|
|
"high": close,
|
|
"low": close,
|
|
"close": close,
|
|
"volume": 1_000.0,
|
|
"amount": 1_000.0 * close,
|
|
}))
|
|
return pd.concat(frames, ignore_index=True)
|
|
|
|
|
|
def test_compute_alpha_schema_and_naming():
|
|
alpha = compute_alpha(_make_data(), "rev5", "reversal", lookback=5)
|
|
assert list(alpha.columns) == ALPHA_COLUMNS
|
|
assert (alpha["alpha_name"] == "rev5").all()
|
|
|
|
|
|
def test_reversal_sign_matches_negative_trailing_return():
|
|
# Cross-sectional z-score preserves the sign relative to the cross-section,
|
|
# so the stock with the most negative trailing return ranks highest.
|
|
data = _make_data()
|
|
alpha = compute_alpha(data, "rev5", "reversal", lookback=5)
|
|
close = data.pivot_table(index="date", columns="symbol_id", values="close").sort_index()
|
|
raw = -close.pct_change(5, fill_method=None)
|
|
last = raw.index[-1]
|
|
expected_top = raw.loc[last].idxmax()
|
|
got = alpha[alpha["date"] == last].set_index("symbol_id")["weight"].idxmax()
|
|
assert got == expected_top
|
|
|
|
|
|
def test_weights_are_cross_sectional_zscore():
|
|
# Each date's weights are a z-score, so the per-date mean is ~0.
|
|
alpha = compute_alpha(_make_data(), "rev5", "reversal", lookback=5)
|
|
per_date_mean = alpha.groupby("date")["weight"].mean().abs()
|
|
assert (per_date_mean < 1e-9).all()
|
|
|
|
|
|
def test_evaluate_alpha_keys():
|
|
data = _make_data()
|
|
alpha = compute_alpha(data, "rev5", "reversal", lookback=5)
|
|
metrics = evaluate_alpha(alpha, data)
|
|
for key in ("cumulative_return", "sharpe_annual", "turnover_annual",
|
|
"max_drawdown", "hit_rate", "n_dates"):
|
|
assert key in metrics
|
|
|
|
|
|
def test_evaluate_alpha_uses_next_open_to_next_open_returns():
|
|
dates = pd.date_range("2024-01-01", periods=5)
|
|
data = pd.concat([
|
|
pd.DataFrame({
|
|
"symbol_id": "sh600000",
|
|
"symbol_name": "sh600000",
|
|
"date": dates,
|
|
"open": [100.0, 100.0, 100.0, 100.0, 200.0],
|
|
"high": [100.0, 1000.0, 1000.0, 1000.0, 1000.0],
|
|
"low": [100.0, 1000.0, 1000.0, 1000.0, 1000.0],
|
|
"close": [100.0, 1000.0, 1000.0, 1000.0, 1000.0],
|
|
"volume": 1_000.0,
|
|
"amount": 1_000.0,
|
|
}),
|
|
pd.DataFrame({
|
|
"symbol_id": "sz000001",
|
|
"symbol_name": "sz000001",
|
|
"date": dates,
|
|
"open": [100.0, 100.0, 100.0, 200.0, 200.0],
|
|
"high": [100.0, 10.0, 10.0, 10.0, 10.0],
|
|
"low": [100.0, 10.0, 10.0, 10.0, 10.0],
|
|
"close": [100.0, 10.0, 10.0, 10.0, 10.0],
|
|
"volume": 1_000.0,
|
|
"amount": 1_000.0,
|
|
}),
|
|
], ignore_index=True)
|
|
alpha = pd.DataFrame({
|
|
"symbol_id": ["sh600000", "sz000001", "sh600000", "sz000001"],
|
|
"date": [dates[1], dates[1], dates[2], dates[2]],
|
|
"alpha_name": ["toy"] * 4,
|
|
"weight": [-1.0, 1.0, 1.0, -1.0],
|
|
})
|
|
|
|
metrics = evaluate_alpha(alpha, data)
|
|
|
|
assert metrics["n_dates"] == 2
|
|
assert np.isclose(metrics["cumulative_return"], 1.25)
|
|
|
|
|
|
def test_evaluate_alpha_excludes_signal_without_forward_return():
|
|
dates = pd.date_range("2024-01-01", periods=3)
|
|
data = pd.concat([
|
|
pd.DataFrame({
|
|
"symbol_id": "sh600000",
|
|
"symbol_name": "sh600000",
|
|
"date": dates,
|
|
"open": [100.0, 100.0, 200.0],
|
|
"high": [100.0, 100.0, 200.0],
|
|
"low": [100.0, 100.0, 200.0],
|
|
"close": [100.0, 100.0, 200.0],
|
|
"volume": 1_000.0,
|
|
"amount": 1_000.0,
|
|
}),
|
|
pd.DataFrame({
|
|
"symbol_id": "sz000001",
|
|
"symbol_name": "sz000001",
|
|
"date": dates,
|
|
"open": [100.0, 100.0, 100.0],
|
|
"high": [100.0, 100.0, 100.0],
|
|
"low": [100.0, 100.0, 100.0],
|
|
"close": [100.0, 100.0, 100.0],
|
|
"volume": 1_000.0,
|
|
"amount": 1_000.0,
|
|
}),
|
|
], ignore_index=True)
|
|
alpha = pd.DataFrame({
|
|
"symbol_id": ["sh600000", "sz000001", "sh600000", "sz000001"],
|
|
"date": [dates[0], dates[0], dates[1], dates[1]],
|
|
"alpha_name": ["toy"] * 4,
|
|
"weight": [1.0, -1.0, -1.0, 1.0],
|
|
})
|
|
|
|
metrics = evaluate_alpha(alpha, data)
|
|
|
|
assert metrics["n_dates"] == 1
|
|
|
|
|
|
def test_equal_weight_is_mean_of_alphas():
|
|
data = _make_data()
|
|
a = compute_alpha(data, "rev", "reversal", lookback=5)
|
|
b = compute_alpha(data, "mom", "momentum", lookback=5)
|
|
combo = _equal_weight([a, b])
|
|
# reversal = -momentum before z-scoring, but after independent per-date
|
|
# z-scoring they are exact negatives, so the equal-weight mean is ~0.
|
|
assert combo["weight"].abs().max() < 1e-9
|
|
|
|
|
|
def test_combine_alphas_schema(tmp_path):
|
|
data = _make_data()
|
|
a_path = tmp_path / "a.pq"
|
|
b_path = tmp_path / "b.pq"
|
|
compute_alpha(data, "rev", "reversal", lookback=5).to_parquet(a_path, index=False)
|
|
compute_alpha(data, "revvol", "reversal_vol", lookback=5, vol_window=10).to_parquet(b_path, index=False)
|
|
combo = combine_alphas([str(a_path), str(b_path)], "eq", method="equal_weight")
|
|
assert list(combo.columns) == COMBO_COLUMNS
|
|
assert (combo["combo_name"] == "eq").all()
|
|
|
|
|
|
def test_combine_single_alpha_is_identity(tmp_path):
|
|
data = _make_data()
|
|
a = compute_alpha(data, "rev", "reversal", lookback=5)
|
|
a_path = tmp_path / "a.pq"
|
|
a.to_parquet(a_path, index=False)
|
|
|
|
combo = combine_alphas([str(a_path)], "rev_combo", method="equal_weight")
|
|
|
|
expected = a[["symbol_id", "date", "weight"]].reset_index(drop=True)
|
|
got = combo[["symbol_id", "date", "weight"]].reset_index(drop=True)
|
|
pd.testing.assert_frame_equal(got, expected)
|
|
assert list(combo.columns) == COMBO_COLUMNS
|
|
assert (combo["combo_name"] == "rev_combo").all()
|
|
|
|
|
|
def test_combine_alphas_rejects_unknown_method(tmp_path):
|
|
data = _make_data()
|
|
alpha_path = tmp_path / "alpha.pq"
|
|
compute_alpha(data, "rev", "reversal", lookback=5).to_parquet(alpha_path, index=False)
|
|
|
|
with pytest.raises(ValueError, match="Unknown combo method"):
|
|
combine_alphas([str(alpha_path)], "bad_combo", method="does_not_exist")
|
|
|
|
|
|
# --- registry / factory -----------------------------------------------------
|
|
|
|
def test_builtins_are_registered():
|
|
assert {"reversal", "reversal_vol", "momentum", "reversal_rank"} <= set(available_alphas())
|
|
|
|
|
|
def test_get_alpha_filters_unaccepted_params():
|
|
# reversal only accepts lookback; passing vol_window too must not error.
|
|
alpha = get_alpha("reversal", lookback=7, vol_window=99)
|
|
assert alpha.name == "reversal"
|
|
assert alpha.lookback == 7
|
|
assert not hasattr(alpha, "vol_window")
|
|
|
|
|
|
def test_get_alpha_forwards_kwargs_to_flexible_alpha():
|
|
@register_alpha
|
|
class _FlexibleAlpha(BaseAlpha):
|
|
name = "_flexible_alpha_kwargs"
|
|
|
|
def __init__(self, **kwargs):
|
|
self.kwargs = kwargs
|
|
|
|
def signal(self, close):
|
|
return close
|
|
|
|
alpha = get_alpha("_flexible_alpha_kwargs", decay=0.5, label="demo")
|
|
|
|
assert alpha.kwargs == {"decay": 0.5, "label": "demo"}
|
|
|
|
|
|
def test_get_alpha_unknown_raises():
|
|
with pytest.raises(KeyError):
|
|
get_alpha("does_not_exist")
|
|
|
|
|
|
def test_register_duplicate_name_raises():
|
|
available_alphas() # ensure built-ins loaded
|
|
|
|
with pytest.raises(ValueError):
|
|
@register_alpha
|
|
class Dup(BaseAlpha):
|
|
name = "reversal"
|
|
|
|
def signal(self, close):
|
|
return close
|
|
|
|
|
|
def test_register_rejects_non_basealpha():
|
|
with pytest.raises(TypeError):
|
|
register_alpha(object) # type: ignore[arg-type]
|
|
|
|
|
|
def test_register_rejects_empty_alpha_name():
|
|
with pytest.raises(ValueError, match="non-empty"):
|
|
@register_alpha
|
|
class NoNameAlpha(BaseAlpha):
|
|
def signal(self, close):
|
|
return close
|
|
|
|
|
|
def test_load_alpha_module_error_paths(tmp_path, monkeypatch):
|
|
missing_path = tmp_path / "missing_alpha.py"
|
|
with pytest.raises(FileNotFoundError):
|
|
load_alpha_module(str(missing_path))
|
|
|
|
bad_path = tmp_path / "bad_alpha.py"
|
|
bad_path.write_text("x = 1\n")
|
|
monkeypatch.setattr(
|
|
"pipeline.alpha.registry.importlib.util.spec_from_file_location",
|
|
lambda *args, **kwargs: None,
|
|
)
|
|
with pytest.raises(ImportError, match="Cannot load alpha module"):
|
|
load_alpha_module(str(bad_path))
|
|
|
|
load_alpha_module("math")
|
|
|
|
|
|
# --- base class --------------------------------------------------------------
|
|
|
|
def test_to_weights_are_per_date_zscore():
|
|
class _Const(BaseAlpha):
|
|
name = "_const_test"
|
|
|
|
def signal(self, close):
|
|
return close # arbitrary finite signal
|
|
|
|
close = _make_data().pivot_table(index="date", columns="symbol_id", values="close")
|
|
weights = _Const().weights(close.sort_index())
|
|
# Each date demeaned to ~0.
|
|
assert (weights.mean(axis=1).abs() < 1e-9).all()
|
|
|
|
|
|
def test_base_alpha_default_signal_and_repr():
|
|
alpha = BaseAlpha()
|
|
alpha.example = 3
|
|
|
|
with pytest.raises(NotImplementedError, match="signal"):
|
|
alpha.signal(pd.DataFrame({"x": [1.0]}))
|
|
with pytest.raises(NotImplementedError, match="signal"):
|
|
alpha.signal_from_data(
|
|
pd.DataFrame({"symbol_id": ["sh600000"]}),
|
|
pd.DataFrame({"sh600000": [1.0]}),
|
|
)
|
|
assert repr(alpha) == "BaseAlpha(example=3)"
|
|
|
|
|
|
# --- external plugin loading -------------------------------------------------
|
|
|
|
def test_load_external_alpha_module(tmp_path):
|
|
module_path = tmp_path / "my_external_alpha.py"
|
|
module_path.write_text(textwrap.dedent('''
|
|
import pandas as pd
|
|
from pipeline.alpha.base import BaseAlpha
|
|
from pipeline.alpha.registry import register_alpha
|
|
|
|
@register_alpha
|
|
class ExternalDemoAlpha(BaseAlpha):
|
|
name = "external_demo"
|
|
|
|
def __init__(self, span: int = 3):
|
|
self.span = span
|
|
|
|
def signal(self, close: pd.DataFrame) -> pd.DataFrame:
|
|
return -close.pct_change(self.span, fill_method=None)
|
|
'''))
|
|
|
|
load_alpha_module(str(module_path))
|
|
assert "external_demo" in available_alphas()
|
|
|
|
# The factory forwards the external alpha's own param (`span`).
|
|
instance = get_alpha("external_demo", span=4, lookback=99)
|
|
assert instance.span == 4
|
|
|
|
# And it works end-to-end through compute_alpha.
|
|
result = compute_alpha(_make_data(), "ext", "external_demo", span=4)
|
|
assert list(result.columns) == ALPHA_COLUMNS
|
|
assert (result["alpha_name"] == "ext").all()
|
|
|
|
|
|
# --- rank reversal + investable universe filter ------------------------------
|
|
|
|
def _make_rich_data(n_days: int = 70, symbols=("sh600000", "sz000001", "sh600519", "sz300750")):
|
|
"""Long-format data with the columns the universe filter needs."""
|
|
dates = pd.date_range("2024-01-01", periods=n_days)
|
|
rng = np.random.default_rng(1)
|
|
frames = []
|
|
for i, sym in enumerate(symbols):
|
|
close = 100.0 + i * 5 + np.cumsum(rng.standard_normal(n_days))
|
|
frames.append(pd.DataFrame({
|
|
"symbol_id": sym,
|
|
"symbol_name": sym,
|
|
"date": dates,
|
|
"open": close, "high": close, "low": close, "close": close,
|
|
"volume": 1_000.0,
|
|
"amount": (1_000.0 + i * 5_000.0) * close, # higher i = more liquid
|
|
"isST": 0,
|
|
"tradestatus": 1,
|
|
}))
|
|
return pd.concat(frames, ignore_index=True)
|
|
|
|
|
|
def test_reversal_rank_registered_and_bounded():
|
|
data = _make_data(n_days=30)
|
|
alpha = compute_alpha(data, "rr", "reversal_rank", lookback=5)
|
|
assert list(alpha.columns) == ALPHA_COLUMNS
|
|
# Rank-demeaned weights are per-date zero-mean and bounded by the
|
|
# cross-section size, never blowing up the way a z-score outlier can.
|
|
per_date_mean = alpha.groupby("date")["weight"].mean().abs()
|
|
assert (per_date_mean < 1e-9).all()
|
|
assert alpha["weight"].abs().max() <= len(data["symbol_id"].unique())
|
|
|
|
|
|
def test_investable_universe_mask_excludes_st_and_suspended():
|
|
data = _make_rich_data()
|
|
# Flag one name ST throughout, and suspend another on the last date.
|
|
data.loc[data["symbol_id"] == "sh600000", "isST"] = 1
|
|
last = data["date"].max()
|
|
data.loc[(data["symbol_id"] == "sz000001") & (data["date"] == last), "tradestatus"] = 0
|
|
|
|
close = data.pivot_table(index="date", columns="symbol_id", values="close").sort_index()
|
|
mask = investable_universe_mask(data, close, top_n=10, min_history=5)
|
|
|
|
assert not mask["sh600000"].any() # ST excluded on every date
|
|
assert not bool(mask.loc[last, "sz000001"]) # suspended on the last date
|
|
assert bool(mask.loc[last, "sh600519"]) # a normal name stays investable
|
|
|
|
|
|
def test_compute_alpha_universe_filter_zeros_excluded_names():
|
|
data = _make_rich_data()
|
|
data.loc[data["symbol_id"] == "sh600000", "isST"] = 1
|
|
|
|
alpha = compute_alpha(
|
|
data, "rr_liq", "reversal_rank", lookback=5,
|
|
universe={"top_n": 10, "min_history": 5},
|
|
)
|
|
# The ST name is never held; an investable name is.
|
|
st_w = alpha.loc[alpha["symbol_id"] == "sh600000", "weight"]
|
|
assert (st_w.fillna(0.0) == 0.0).all()
|
|
assert alpha.loc[alpha["symbol_id"] == "sz300750", "weight"].abs().sum() > 0.0
|
|
|
|
|
|
def test_universe_filter_does_not_corrupt_signal_history():
|
|
# Masking happens on the signal, not the price history, so weights on
|
|
# investable names match the unfiltered weights restricted to that set.
|
|
data = _make_rich_data()
|
|
universe = {"top_n": 2, "min_history": 5} # keep only the 2 most liquid names
|
|
|
|
filtered = compute_alpha(data, "f", "reversal_rank", lookback=5, universe=universe)
|
|
held = set(filtered.loc[filtered["weight"] != 0.0, "symbol_id"].unique())
|
|
# The two most liquid names (highest amount) are sh600519, sz300750.
|
|
assert held == {"sh600519", "sz300750"}
|
|
|
|
|
|
# --- feature-aware alpha integration ----------------------------------------
|
|
|
|
def test_compute_alpha_without_feature_path_matches_empty_feature_paths():
|
|
data = _make_data()
|
|
|
|
base = compute_alpha(data, "rev5", "reversal", lookback=5)
|
|
with_empty_features = compute_alpha(
|
|
data,
|
|
"rev5",
|
|
"reversal",
|
|
lookback=5,
|
|
feature_paths=[],
|
|
)
|
|
|
|
pd.testing.assert_frame_equal(base, with_empty_features)
|
|
|
|
|
|
def test_feature_aware_alpha_reads_joined_feature_column(tmp_path):
|
|
module_path = tmp_path / "feature_aware_alpha.py"
|
|
module_path.write_text(textwrap.dedent('''
|
|
import pandas as pd
|
|
from pipeline.alpha.base import BaseAlpha
|
|
from pipeline.alpha.registry import register_alpha
|
|
|
|
@register_alpha
|
|
class FeatureAwareAlpha(BaseAlpha):
|
|
name = "feature_aware_test_alpha"
|
|
|
|
def signal_from_data(
|
|
self,
|
|
data: pd.DataFrame,
|
|
close: pd.DataFrame,
|
|
) -> pd.DataFrame:
|
|
signal = data.pivot_table(
|
|
index="date",
|
|
columns="symbol_id",
|
|
values="toy_feature",
|
|
aggfunc="first",
|
|
)
|
|
return signal.reindex(index=close.index, columns=close.columns)
|
|
'''))
|
|
|
|
data = _make_data()
|
|
feature = data[["symbol_id", "date"]].copy()
|
|
feature["toy_feature"] = feature["symbol_id"].map({
|
|
"sh600000": 1.0,
|
|
"sz000001": 2.0,
|
|
"sh600519": 3.0,
|
|
})
|
|
feature_path = tmp_path / "toy_feature.pq"
|
|
feature.to_parquet(feature_path, index=False)
|
|
|
|
load_alpha_module(str(module_path))
|
|
result = compute_alpha(
|
|
data,
|
|
"feature_run",
|
|
"feature_aware_test_alpha",
|
|
feature_paths=[str(feature_path)],
|
|
)
|
|
|
|
assert list(result.columns) == ALPHA_COLUMNS
|
|
assert (result["alpha_name"] == "feature_run").all()
|
|
last = result[result["date"] == result["date"].max()]
|
|
assert last.set_index("symbol_id")["weight"].idxmax() == "sh600519"
|
|
|
|
|
|
def test_feature_paths_join_multiple_files_and_normalize_dates(tmp_path):
|
|
module_path = tmp_path / "multi_feature_alpha.py"
|
|
module_path.write_text(textwrap.dedent('''
|
|
import pandas as pd
|
|
from pipeline.alpha.base import BaseAlpha
|
|
from pipeline.alpha.registry import register_alpha
|
|
|
|
@register_alpha
|
|
class MultiFeatureAlpha(BaseAlpha):
|
|
name = "multi_feature_test_alpha"
|
|
|
|
def signal_from_data(
|
|
self,
|
|
data: pd.DataFrame,
|
|
close: pd.DataFrame,
|
|
) -> pd.DataFrame:
|
|
data = data.copy()
|
|
data["combined_feature"] = data["toy_a"] + data["toy_b"]
|
|
signal = data.pivot_table(
|
|
index="date",
|
|
columns="symbol_id",
|
|
values="combined_feature",
|
|
aggfunc="first",
|
|
)
|
|
return signal.reindex(index=close.index, columns=close.columns)
|
|
'''))
|
|
|
|
data = _make_data(n_days=8)
|
|
symbol_score = {"sh600000": 1.0, "sz000001": 2.0, "sh600519": 3.0}
|
|
|
|
feature_a = data[["symbol_id", "date"]].copy()
|
|
feature_a["date"] = feature_a["date"] + pd.Timedelta(hours=15)
|
|
feature_a["toy_a"] = feature_a["symbol_id"].map(symbol_score)
|
|
|
|
feature_b = data[["symbol_id", "date"]].copy()
|
|
feature_b["date"] = feature_b["date"].dt.strftime("%Y-%m-%d 09:30:00")
|
|
feature_b["toy_b"] = feature_b["symbol_id"].map(symbol_score) * 10.0
|
|
|
|
feature_a_path = tmp_path / "toy_a.pq"
|
|
feature_b_path = tmp_path / "toy_b.pq"
|
|
feature_a.to_parquet(feature_a_path, index=False)
|
|
feature_b.to_parquet(feature_b_path, index=False)
|
|
|
|
load_alpha_module(str(module_path))
|
|
result = compute_alpha(
|
|
data,
|
|
"multi_feature_run",
|
|
"multi_feature_test_alpha",
|
|
feature_paths=[str(feature_a_path), str(feature_b_path)],
|
|
)
|
|
|
|
assert list(result.columns) == ALPHA_COLUMNS
|
|
assert (result["alpha_name"] == "multi_feature_run").all()
|
|
last = result[result["date"] == result["date"].max()]
|
|
assert last.set_index("symbol_id")["weight"].idxmax() == "sh600519"
|
|
|
|
|
|
def test_compute_alpha_rejects_duplicate_feature_frame_columns():
|
|
data = _make_data()
|
|
duplicate_columns = pd.DataFrame(
|
|
[["sh600000", pd.Timestamp("2024-01-01"), 1.0, 2.0]],
|
|
columns=["symbol_id", "date", "toy_feature", "toy_feature"],
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="duplicate columns"):
|
|
compute_alpha(
|
|
data,
|
|
"bad_features",
|
|
"reversal",
|
|
feature_frames=[duplicate_columns],
|
|
)
|
|
|
|
|
|
def test_compute_alpha_rejects_feature_path_collision_with_daily_data(tmp_path):
|
|
data = _make_data()
|
|
close_collision = data[["symbol_id", "date"]].copy()
|
|
close_collision["close"] = 1.0
|
|
close_collision_path = tmp_path / "close_collision.pq"
|
|
close_collision.to_parquet(close_collision_path, index=False)
|
|
|
|
with pytest.raises(ValueError, match="conflict"):
|
|
compute_alpha(
|
|
data,
|
|
"close_collision",
|
|
"reversal",
|
|
feature_paths=[str(close_collision_path)],
|
|
)
|
|
|
|
|
|
def test_evaluate_alpha_empty_when_signal_dates_not_on_market_calendar():
|
|
data = _make_data(n_days=3)
|
|
alpha = pd.DataFrame({
|
|
"symbol_id": ["sh600000"],
|
|
"date": [pd.Timestamp("2030-01-01")],
|
|
"alpha_name": ["future"],
|
|
"weight": [1.0],
|
|
})
|
|
|
|
metrics = evaluate_alpha(alpha, data)
|
|
|
|
assert metrics["n_dates"] == 0
|
|
assert metrics["cumulative_return"] == 0.0
|