gahow/quant-trading
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: remove pre-lab research and dead strategies; index them in research/ARCHIVE.md

Browse Source 
Removed (79 files, all in history; research/ARCHIVE.md lists each with its
purpose and the commit to restore it from):
- root factor_*.py mining scripts (2026-04) and yearly_sweep.py
- scripts/run_daily.{py,sh} (superseded by the monitor service; no crontab)
- strategies no registry, test or live book uses: trend_rider_v8,
  adaptive_blend, cross_asset_momentum, long_hedged, ls_momentum
- 55 research/*.py scripts from 2026-04/05 written before research/lab,
  the us-alpha design spec, and the 7 tests that only covered them

Kept research/permanent_yearly.py: bridge.py and precompute_backtest.py
import equity_with_cashflows. CLAUDE.md gains the archive note and the
command lines of the studies committed in the previous commit.

Checks: 99 unit tests OK (125 before, minus the 26 removed with their
modules), all 157 remaining modules import, 94 registry strategies build,
bridge.py list OK, web 23 tests OK and build OK.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Gahow Wang
2026-10-09 22:20:59 -07:00
parent 04e2c5c3be
commit a9683b6d1f
81 changed files with 121 additions and 19668 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
tests/test_alpha_signals.py
View File

@@ -1,118 +0,0 @@
import unittest
import warnings
import numpy as np
import pandas as pd
class AlphaSignalTests(unittest.TestCase):
def test_build_regime_filter_requires_market_trend_and_non_market_leader(self):
from research.regime_filters import build_regime_filter
dates = pd.date_range("2023-01-01", periods=260, freq="D")
spy = pd.Series([100.0 + i for i in range(260)], index=dates)
qqq_leader = pd.Series([100.0 + 1.4 * i for i in range(260)], index=dates)
xlu = pd.Series([100.0 + 0.2 * i for i in range(260)], index=dates)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
bullish = build_regime_filter(pd.DataFrame({"SPY": spy, "QQQ": qqq_leader, "XLU": xlu}))
qqq_laggard = pd.Series([100.0 + 0.5 * i for i in range(260)], index=dates)
no_leader = build_regime_filter(pd.DataFrame({"SPY": spy, "QQQ": qqq_laggard, "XLU": xlu}))
self.assertEqual(len(caught), 0)
self.assertFalse(bool(bullish.iloc[199]))
self.assertTrue(bool(bullish.iloc[-1]))
self.assertFalse(bool(no_leader.iloc[-1]))
def test_build_regime_filter_handles_internal_missing_prices_without_warnings(self):
from research.regime_filters import build_regime_filter
dates = pd.date_range("2023-01-01", periods=260, freq="D")
spy = pd.Series([100.0 + i for i in range(260)], index=dates)
qqq = pd.Series([100.0 + 1.4 * i for i in range(260)], index=dates)
qqq.iloc[120] = np.nan
etf_close = pd.DataFrame({"SPY": spy, "QQQ": qqq, "XLU": 100.0}, index=dates)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
regime = build_regime_filter(etf_close)
self.assertEqual(len(caught), 0)
self.assertEqual(str(regime.dtype), "bool")
def test_breakout_after_compression_score_is_shifted_and_rewards_breakout_profile(self):
from research.event_factors import breakout_after_compression_score
dates = pd.date_range("2024-01-01", periods=80, freq="D")
aaa_close = [100.0 + i for i in range(60)] + [159.0 + 0.05 * i for i in range(20)]
bbb_close = [100.0 + i for i in range(60)] + [150.0 - i for i in range(20)]
close = pd.DataFrame({"AAA": aaa_close, "BBB": bbb_close}, index=dates)
high = pd.DataFrame(
{
"AAA": [value + 0.4 for value in aaa_close],
"BBB": [value + 4.0 for value in bbb_close],
},
index=dates,
)
low = pd.DataFrame(
{
"AAA": [value - 0.4 for value in aaa_close],
"BBB": [value - 4.0 for value in bbb_close],
},
index=dates,
)
volume = pd.DataFrame(
{
"AAA": [1_000.0] * 79 + [1_000.0],
"BBB": [1_000.0] * 80,
},
index=dates,
)
volume.loc[dates[-2], "AAA"] = 6_000.0
shifted_result = breakout_after_compression_score(close, high, low, volume)
self.assertGreater(
shifted_result.loc[dates[-1], "AAA"],
shifted_result.loc[dates[-1], "BBB"],
)
changed_last_day = close.copy()
changed_last_day_high = high.copy()
changed_last_day_low = low.copy()
changed_last_day_volume = volume.copy()
changed_last_day.loc[dates[-1], "AAA"] = 120.0
changed_last_day_high.loc[dates[-1], "AAA"] = 130.0
changed_last_day_low.loc[dates[-1], "AAA"] = 110.0
changed_last_day_volume.loc[dates[-1], "AAA"] = 20_000.0
last_day_changed_result = breakout_after_compression_score(
changed_last_day,
changed_last_day_high,
changed_last_day_low,
changed_last_day_volume,
)
self.assertEqual(
shifted_result.loc[dates[-1], "AAA"],
last_day_changed_result.loc[dates[-1], "AAA"],
)
def test_breakout_after_compression_score_keeps_float_output_when_denominators_hit_zero(self):
from research.event_factors import breakout_after_compression_score
dates = pd.date_range("2024-01-01", periods=70, freq="D")
close = pd.DataFrame({"AAA": [10.0] * 70}, index=dates)
high = pd.DataFrame({"AAA": [10.0] * 70}, index=dates)
low = pd.DataFrame({"AAA": [10.0] * 70}, index=dates)
volume = pd.DataFrame({"AAA": [0.0] * 70}, index=dates)
score = breakout_after_compression_score(close, high, low, volume)
self.assertEqual(str(score.dtypes["AAA"]), "float64")
self.assertTrue(pd.isna(score.iloc[-1]["AAA"]))
if __name__ == "__main__":
unittest.main()

46
tests/test_fetch_historical.py
View File

@@ -1,46 +0,0 @@
import tempfile
import unittest
from pathlib import Path
from unittest import mock
import pandas as pd
from research import fetch_historical
class FetchHistoricalTests(unittest.TestCase):
def test_fetch_all_historical_ohlcv_writes_field_specific_csvs(self):
dates = pd.to_datetime(["2024-01-02", "2024-01-03"])
raw = pd.DataFrame(
{
("Close", "AAA"): [10.0, 11.0],
("Close", "BBB"): [20.0, 21.0],
("High", "AAA"): [10.5, 11.5],
("High", "BBB"): [20.5, 21.5],
("Low", "AAA"): [9.5, 10.5],
("Low", "BBB"): [19.5, 20.5],
("Volume", "AAA"): [1000.0, 1100.0],
("Volume", "BBB"): [2000.0, 2100.0],
},
index=dates,
)
raw.columns = pd.MultiIndex.from_tuples(raw.columns)
with tempfile.TemporaryDirectory() as tmpdir:
with mock.patch.object(fetch_historical, "DATA_DIR", tmpdir):
with mock.patch.object(fetch_historical, "OUT_PATH", str(Path(tmpdir) / "us_pit.csv")):
with mock.patch("research.fetch_historical.uh.load_sp500_history", return_value={"AAA": [[None, None]], "BBB": [[None, None]]}):
with mock.patch("research.fetch_historical.uh.all_tickers_ever", return_value=["AAA", "BBB"]):
with mock.patch("research.fetch_historical.yf.download", return_value=raw):
panels = fetch_historical.fetch_all_historical_ohlcv(force=True)
self.assertEqual(set(panels.keys()), {"close", "high", "low", "volume"})
self.assertTrue((Path(tmpdir) / "us_pit.csv").exists())
self.assertTrue((Path(tmpdir) / "us_pit_close.csv").exists())
self.assertTrue((Path(tmpdir) / "us_pit_high.csv").exists())
self.assertTrue((Path(tmpdir) / "us_pit_low.csv").exists())
self.assertTrue((Path(tmpdir) / "us_pit_volume.csv").exists())
if __name__ == "__main__":
unittest.main()

91
tests/test_trend_rider_robustness.py
View File

@@ -1,91 +0,0 @@
import unittest
import numpy as np
import pandas as pd
from research import trend_rider_robustness as robustness
class TrendRiderRobustnessTests(unittest.TestCase):
def test_evaluate_weights_reports_core_risk_metrics(self):
dates = pd.bdate_range("2024-01-01", periods=6)
prices = pd.DataFrame(
{
"AAA": [100, 110, 105, 120, 118, 130],
"BBB": [50, 49, 51, 50, 52, 53],
},
index=dates,
)
weights = pd.DataFrame(
{
"AAA": [0, 1, 1, 0, 0, 1],
"BBB": [0, 0, 0, 1, 1, 0],
},
index=dates,
)
result = robustness.evaluate_weights("synthetic", weights, prices, transaction_cost=0.001)
self.assertEqual(result.name, "synthetic")
self.assertGreater(result.final_multiple, 1.0)
self.assertLessEqual(result.max_drawdown, 0.0)
self.assertGreater(result.switches, 0)
self.assertGreater(result.avg_daily_turnover, 0.0)
def test_parameter_sweep_returns_rankable_rows(self):
dates = pd.bdate_range("2023-01-02", periods=320)
trend = np.linspace(100, 180, len(dates))
prices = pd.DataFrame(
{
"SPY": trend,
"TQQQ": trend * 1.5,
"UPRO": trend * 1.4,
"GLD": np.linspace(100, 105, len(dates)),
"DBC": np.linspace(90, 95, len(dates)),
},
index=dates,
)
sweep = robustness.parameter_sweep(
prices,
variants=[
{"vol_enter": 0.14, "dd_stop": 0.05, "peak_enter": 0.02, "mom_lookback": 63},
{"vol_enter": 0.16, "dd_stop": 0.07, "peak_enter": 0.03, "mom_lookback": 84},
],
start="2023-01-02",
)
self.assertEqual(len(sweep), 2)
self.assertIn("cagr", sweep.columns)
self.assertIn("max_drawdown", sweep.columns)
self.assertTrue(sweep["cagr"].notna().all())
def test_candidate_weights_include_v4_and_market_benchmarks(self):
dates = pd.bdate_range("2023-01-02", periods=320)
trend = np.linspace(100, 180, len(dates))
prices = pd.DataFrame(
{
"SPY": trend,
"QQQ": trend * 1.1,
"SSO": trend * 1.5,
"QLD": trend * 1.6,
"UPRO": trend * 2.0,
"TQQQ": trend * 2.2,
"SHY": np.linspace(100, 103, len(dates)),
"IEF": np.linspace(100, 104, len(dates)),
"TLT": np.linspace(100, 105, len(dates)),
"GLD": np.linspace(100, 115, len(dates)),
"DBC": np.linspace(90, 105, len(dates)),
},
index=dates,
)
candidates = robustness.candidate_weights(prices)
self.assertIn("TrendRiderV4", candidates)
self.assertIn("SPY Buy&Hold", candidates)
self.assertIn("QQQ Buy&Hold", candidates)
if __name__ == "__main__":
unittest.main()

276
tests/test_us_alpha_pipeline.py
View File

@@ -1,276 +0,0 @@
import unittest
from pathlib import Path
from unittest import mock
import pandas as pd
class USAlphaPipelineTests(unittest.TestCase):
def test_build_equal_weight_portfolio_caps_holdings_under_ties(self):
from research.us_alpha_pipeline import _build_equal_weight_portfolio
dates = pd.date_range("2024-01-01", periods=2, freq="D")
score = pd.DataFrame(
{
"AAA": [0.9, 0.9],
"BBB": [0.9, 0.9],
"CCC": [0.9, 0.9],
},
index=dates,
)
tradable_mask = pd.DataFrame(True, index=dates, columns=score.columns)
regime = pd.Series([True, True], index=dates)
weights = _build_equal_weight_portfolio(score, tradable_mask, regime, top_n=2)
self.assertEqual(int((weights.iloc[-1] > 0).sum()), 2)
self.assertAlmostEqual(float(weights.iloc[-1].sum()), 1.0)
def test_equity_curve_uses_prior_day_weights_for_returns(self):
from research.us_alpha_pipeline import _equity_curve
dates = pd.date_range("2024-01-01", periods=3, freq="D")
close = pd.DataFrame({"AAA": [1.0, 2.0, 4.0]}, index=dates)
weights = pd.DataFrame({"AAA": [0.0, 1.0, 0.0]}, index=dates)
equity = _equity_curve(close, weights)
self.assertEqual(float(equity.iloc[1]), 2.0)
self.assertEqual(float(equity.iloc[2]), 2.0)
def test_summarize_equity_window_returns_nans_when_history_is_too_short(self):
from research.us_alpha_report import summarize_equity_window
dates = pd.date_range("2024-01-01", periods=10, freq="D")
equity = pd.Series([1.0 + 0.01 * i for i in range(10)], index=dates)
summary = summarize_equity_window(equity, "demo", window_years=1)
self.assertTrue(pd.isna(summary["CAGR"]))
self.assertTrue(pd.isna(summary["Sharpe"]))
self.assertTrue(pd.isna(summary["MaxDD"]))
self.assertTrue(pd.isna(summary["TotalRet"]))
def test_run_alpha_pipeline_returns_expected_strategy_summary(self):
from research.us_alpha_pipeline import run_alpha_pipeline
dates = pd.date_range("2023-01-01", periods=400, freq="D")
aaa_close = [50.0 + 0.20 * i for i in range(400)]
bbb_close = [55.0 + 0.12 * i for i in range(400)]
ccc_close = [60.0 + 0.05 * i for i in range(400)]
close = pd.DataFrame(
{
"AAA": aaa_close,
"BBB": bbb_close,
"CCC": ccc_close,
},
index=dates,
)
high = pd.DataFrame(
{
"AAA": [value + 0.5 for value in aaa_close],
"BBB": [value + 1.0 for value in bbb_close],
"CCC": [value + 1.5 for value in ccc_close],
},
index=dates,
)
low = pd.DataFrame(
{
"AAA": [value - 0.5 for value in aaa_close],
"BBB": [value - 1.0 for value in bbb_close],
"CCC": [value - 1.5 for value in ccc_close],
},
index=dates,
)
volume = pd.DataFrame(
{
"AAA": [1_500_000.0] * 400,
"BBB": [1_400_000.0] * 400,
"CCC": [1_300_000.0] * 400,
},
index=dates,
)
volume.loc[dates[-2], "AAA"] = 4_000_000.0
etf_close = pd.DataFrame(
{
"SPY": [300.0 + 0.8 * i for i in range(400)],
"QQQ": [280.0 + 1.1 * i for i in range(400)],
"XLF": [200.0 + 0.4 * i for i in range(400)],
},
index=dates,
)
market_data = {
"close": close,
"high": high,
"low": low,
"volume": volume,
}
summary = run_alpha_pipeline(
market_data=market_data,
etf_close=etf_close,
pit_membership=None,
windows=(1,),
top_n=2,
)
required_columns = {"strategy", "window_years", "CAGR", "Sharpe", "MaxDD", "TotalRet"}
self.assertTrue(required_columns.issubset(summary.columns))
self.assertEqual(set(summary["strategy"]), {"breakout_regime", "rank_blend_regime"})
self.assertEqual(set(summary["window_years"]), {1})
self.assertEqual(len(summary), 2)
self.assertTrue(summary[["CAGR", "Sharpe", "MaxDD", "TotalRet"]].notna().all().all())
def test_run_alpha_pipeline_includes_fundamental_variants_when_score_is_supplied(self):
from research.us_alpha_pipeline import run_alpha_pipeline
dates = pd.date_range("2023-01-01", periods=400, freq="D")
close = pd.DataFrame(
{
"AAA": [50.0 + 0.20 * i for i in range(400)],
"BBB": [55.0 + 0.12 * i for i in range(400)],
"CCC": [60.0 + 0.05 * i for i in range(400)],
},
index=dates,
)
high = close + 1.0
low = close - 1.0
volume = pd.DataFrame(
{
"AAA": [1_500_000.0] * 400,
"BBB": [1_400_000.0] * 400,
"CCC": [1_300_000.0] * 400,
},
index=dates,
)
etf_close = pd.DataFrame(
{
"SPY": [300.0 + 0.8 * i for i in range(400)],
"QQQ": [280.0 + 1.1 * i for i in range(400)],
"XLF": [200.0 + 0.4 * i for i in range(400)],
},
index=dates,
)
market_data = {
"close": close,
"high": high,
"low": low,
"volume": volume,
}
fundamental_score = pd.DataFrame(
{
"AAA": [0.9] * 400,
"BBB": [0.6] * 400,
"CCC": [0.3] * 400,
},
index=dates,
)
summary = run_alpha_pipeline(
market_data=market_data,
etf_close=etf_close,
pit_membership=None,
windows=(1,),
top_n=2,
fundamental_score=fundamental_score,
)
self.assertEqual(
set(summary["strategy"]),
{
"breakout_regime",
"rank_blend_regime",
"fundamental_regime",
"breakout_fundamental_regime",
"rank_blend_fundamental_regime",
},
)
def test_run_alpha_pipeline_close_only_fallback_skips_breakout_and_uses_spy_regime(self):
from research.us_alpha_pipeline import run_alpha_pipeline
dates = pd.date_range("2023-01-01", periods=400, freq="D")
close = pd.DataFrame(
{
"AAA": [50.0 + 0.20 * i for i in range(400)],
"BBB": [55.0 + 0.12 * i for i in range(400)],
"CCC": [60.0 + 0.05 * i for i in range(400)],
},
index=dates,
)
market_data = {
"close": close,
"high": pd.DataFrame(index=dates, columns=close.columns),
"low": pd.DataFrame(index=dates, columns=close.columns),
"volume": pd.DataFrame(index=dates, columns=close.columns),
}
etf_close = pd.DataFrame({"SPY": [300.0 + 0.8 * i for i in range(400)]}, index=dates)
fundamental_score = pd.DataFrame(
{
"AAA": [0.9] * 400,
"BBB": [0.6] * 400,
"CCC": [0.3] * 400,
},
index=dates,
)
summary = run_alpha_pipeline(
market_data=market_data,
etf_close=etf_close,
pit_membership=None,
windows=(1,),
top_n=2,
fundamental_score=fundamental_score,
)
self.assertEqual(
set(summary["strategy"]),
{
"rank_blend_regime",
"fundamental_regime",
"rank_blend_fundamental_regime",
},
)
self.assertTrue(summary[["CAGR", "Sharpe", "MaxDD", "TotalRet"]].notna().all().all())
def test_run_saved_pit_alpha_pipeline_reads_saved_inputs(self):
from research.us_alpha_pipeline import run_saved_pit_alpha_pipeline
dates = pd.date_range("2024-01-01", periods=320, freq="D")
close = pd.DataFrame(
{
"AAA": [50.0 + 0.2 * i for i in range(320)],
"BBB": [40.0 + 0.1 * i for i in range(320)],
},
index=dates,
)
high = close + 1.0
low = close - 1.0
volume = pd.DataFrame({"AAA": [2_500_000.0] * 320, "BBB": [2_000_000.0] * 320}, index=dates)
etf_close = pd.DataFrame(
{"SPY": [300.0 + 0.8 * i for i in range(320)], "QQQ": [280.0 + 1.1 * i for i in range(320)]},
index=dates,
)
with self.subTest("saved_inputs"):
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
close.to_csv(Path(tmpdir) / "us_pit_close.csv")
high.to_csv(Path(tmpdir) / "us_pit_high.csv")
low.to_csv(Path(tmpdir) / "us_pit_low.csv")
volume.to_csv(Path(tmpdir) / "us_pit_volume.csv")
etf_close.to_csv(Path(tmpdir) / "us_etf.csv")
intervals = {"AAA": [[None, None]], "BBB": [[None, None]]}
with mock.patch("research.us_alpha_pipeline.uh.load_sp500_history", return_value=intervals):
summary = run_saved_pit_alpha_pipeline(data_dir=tmpdir, windows=(1,), top_n=1)
self.assertEqual(set(summary["strategy"]), {"breakout_regime", "rank_blend_regime"})
if __name__ == "__main__":
unittest.main()

65
tests/test_us_combo_sweep.py
View File

@@ -1,65 +0,0 @@
import unittest
import pandas as pd
class USComboSweepTests(unittest.TestCase):
def test_apply_filter_threshold_masks_names_below_rank_cutoff(self):
from research.us_combo_sweep import apply_filter_threshold
index = pd.DatetimeIndex([pd.Timestamp("2024-01-31")])
score = pd.DataFrame({"AAA": [0.9], "BBB": [0.8], "CCC": [0.7]}, index=index)
filter_rank = pd.DataFrame({"AAA": [0.2], "BBB": [0.6], "CCC": [0.9]}, index=index)
filtered = apply_filter_threshold(score, filter_rank, min_rank=0.5)
self.assertTrue(pd.isna(filtered.iloc[0]["AAA"]))
self.assertEqual(float(filtered.iloc[0]["BBB"]), 0.8)
self.assertEqual(float(filtered.iloc[0]["CCC"]), 0.7)
def test_run_combo_backtests_returns_candidates_and_yearly_summary(self):
from research.us_combo_sweep import run_combo_backtests
dates = pd.date_range("2022-01-01", periods=800, freq="D")
close = pd.DataFrame(
{
"AAA": [50.0 + 0.12 * i for i in range(800)],
"BBB": [40.0 + 0.08 * i for i in range(800)],
"CCC": [35.0 + 0.06 * i for i in range(800)],
"DDD": [30.0 + 0.04 * i for i in range(800)],
"EEE": [25.0 + 0.03 * i for i in range(800)],
"FFF": [20.0 + 0.02 * i for i in range(800)],
"GGG": [18.0 + 0.015 * i for i in range(800)],
"HHH": [16.0 + 0.010 * i for i in range(800)],
"III": [14.0 + 0.008 * i for i in range(800)],
"JJJ": [12.0 + 0.005 * i for i in range(800)],
"SPY": [300.0 + 0.20 * i for i in range(800)],
},
index=dates,
)
fundamental_score = pd.DataFrame(
{
"AAA": [0.95] * 800,
"BBB": [0.90] * 800,
"CCC": [0.85] * 800,
"DDD": [0.80] * 800,
"EEE": [0.75] * 800,
"FFF": [0.70] * 800,
"GGG": [0.65] * 800,
"HHH": [0.60] * 800,
"III": [0.55] * 800,
"JJJ": [0.50] * 800,
},
index=dates,
)
yearly, summary = run_combo_backtests(close, fundamental_score, top_n=3)
self.assertIn("Recovery+Mom Top10", yearly.columns)
self.assertIn("rm_fund_tilt_20", yearly.columns)
self.assertIn("rm_fund_filter_50", yearly.columns)
self.assertIn("mega_quality_fund", set(summary["strategy"]))
if __name__ == "__main__":
unittest.main()

52
tests/test_us_fundamentals.py
View File

@@ -1,52 +0,0 @@
import unittest
import pandas as pd
class USFundamentalTransformsTests(unittest.TestCase):
def test_build_quarterly_factor_pack_derives_expected_signals(self):
from research.us_fundamentals import build_quarterly_factor_pack
quarter_ends = pd.to_datetime(
["2023-03-31", "2023-06-30", "2023-09-30", "2023-12-31", "2024-03-31"]
)
close = pd.DataFrame(
{"AAA": [100.0, 105.0], "BBB": [50.0, 48.0]},
index=pd.to_datetime(["2024-06-03", "2024-06-04"]),
)
quarterly = {
"net_income": pd.DataFrame(
{"AAA": [10.0, 11.0, 12.0, 13.0, 14.0], "BBB": [4.0, 4.0, 5.0, 5.0, 5.0]},
index=quarter_ends,
),
"gross_profit": pd.DataFrame(
{"AAA": [30.0, 31.0, 32.0, 33.0, 34.0], "BBB": [10.0, 10.0, 11.0, 11.0, 11.0]},
index=quarter_ends,
),
"equity": pd.DataFrame(
{"AAA": [200.0, 205.0, 210.0, 215.0, 220.0], "BBB": [80.0, 81.0, 82.0, 83.0, 84.0]},
index=quarter_ends,
),
"assets": pd.DataFrame(
{"AAA": [300.0, 305.0, 310.0, 315.0, 320.0], "BBB": [130.0, 131.0, 132.0, 133.0, 134.0]},
index=quarter_ends,
),
"shares": pd.DataFrame(
{"AAA": [10.0, 10.0, 10.0, 10.0, 10.0], "BBB": [10.0, 10.0, 11.0, 11.0, 11.0]},
index=quarter_ends,
),
}
factor_pack = build_quarterly_factor_pack(quarterly, close, lag_days=60)
self.assertIn("composite", factor_pack)
self.assertIn("book_to_market", factor_pack)
self.assertEqual(list(factor_pack["composite"].columns), ["AAA", "BBB"])
self.assertGreater(
float(factor_pack["composite"].iloc[-1]["AAA"]),
float(factor_pack["composite"].iloc[-1]["BBB"]),
)
if __name__ == "__main__":
unittest.main()

213
tests/test_us_universe.py
View File

@@ -1,213 +0,0 @@
import unittest
import warnings
import pandas as pd
class BuildTradableMaskTests(unittest.TestCase):
def test_build_tradable_mask_uses_only_lagged_price_and_liquidity_inputs(self):
from research.us_universe import build_tradable_mask
dates = pd.date_range("2024-01-01", periods=4, freq="D")
close = pd.DataFrame({"AAA": [4.0, 10.0, 10.0, 10.0]}, index=dates)
volume = pd.DataFrame({"AAA": [float("nan"), 200.0, 200.0, 200.0]}, index=dates)
mask = build_tradable_mask(
close=close,
volume=volume,
pit_membership=None,
min_price=5.0,
min_dollar_volume=1000.0,
min_history_days=2,
min_valid_volume_days=2,
liquidity_window=2,
)
expected = pd.DataFrame({"AAA": [False, False, False, True]}, index=dates, dtype=bool)
pd.testing.assert_frame_equal(mask, expected)
def test_build_tradable_mask_uses_only_lagged_history(self):
from research.us_universe import build_tradable_mask
dates = pd.date_range("2024-01-01", periods=4, freq="D")
close = pd.DataFrame({"AAA": [10.0, float("nan"), 10.0, 10.0]}, index=dates)
volume = pd.DataFrame({"AAA": [200.0, 200.0, 200.0, 200.0]}, index=dates)
mask = build_tradable_mask(
close=close,
volume=volume,
pit_membership=None,
min_price=5.0,
min_dollar_volume=1_000.0,
min_history_days=2,
min_valid_volume_days=1,
liquidity_window=1,
)
expected = pd.DataFrame({"AAA": [False, False, False, False]}, index=dates, dtype=bool)
pd.testing.assert_frame_equal(mask, expected)
def test_build_tradable_mask_requires_membership_history_before_first_eligible_day(self):
from research.us_universe import build_tradable_mask
dates = pd.date_range("2024-01-01", periods=4, freq="D")
close = pd.DataFrame({"AAA": [10.0, 10.0, 10.0, 10.0]}, index=dates)
volume = pd.DataFrame({"AAA": [200.0, 200.0, 200.0, 200.0]}, index=dates)
pit_membership = pd.DataFrame({"AAA": [False, False, True, True]}, index=dates)
mask = build_tradable_mask(
close=close,
volume=volume,
pit_membership=pit_membership,
min_price=5.0,
min_dollar_volume=1_000.0,
min_history_days=1,
min_valid_volume_days=1,
liquidity_window=1,
)
expected = pd.DataFrame({"AAA": [False, False, False, True]}, index=dates, dtype=bool)
pd.testing.assert_frame_equal(mask, expected)
def test_build_tradable_mask_aligns_pit_membership_without_truthy_carryover(self):
from research.us_universe import build_tradable_mask
dates = pd.date_range("2024-01-01", periods=3, freq="D")
close = pd.DataFrame(
{
"AAA": [10.0, 10.0, 10.0],
"BBB": [12.0, 12.0, 12.0],
},
index=dates,
)
volume = pd.DataFrame(
{
"AAA": [1_000_000.0, 1_000_000.0, 1_000_000.0],
"BBB": [1_000_000.0, 1_000_000.0, 1_000_000.0],
},
index=dates,
)
pit_membership = pd.DataFrame(
{
"BBB": [True, True, False],
"CCC": [True, True, True],
},
index=pd.date_range("2024-01-02", periods=3, freq="D"),
)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
mask = build_tradable_mask(
close=close,
volume=volume,
pit_membership=pit_membership,
min_price=5.0,
min_dollar_volume=1_000.0,
min_history_days=1,
min_valid_volume_days=1,
liquidity_window=1,
)
self.assertEqual(len(caught), 0)
expected = pd.DataFrame(
{
"AAA": [False, False, False],
"BBB": [False, False, True],
},
index=dates,
dtype=bool,
)
pd.testing.assert_frame_equal(mask, expected)
def test_build_tradable_mask_treats_missing_membership_cells_as_false(self):
from research.us_universe import build_tradable_mask
dates = pd.date_range("2024-01-01", periods=3, freq="D")
close = pd.DataFrame({"AAA": [10.0, 10.0, 10.0]}, index=dates)
volume = pd.DataFrame({"AAA": [1_000_000.0, 1_000_000.0, 1_000_000.0]}, index=dates)
pit_membership = pd.DataFrame(
{"AAA": [True, pd.NA, True]},
index=dates,
dtype="boolean",
)
mask = build_tradable_mask(
close=close,
volume=volume,
pit_membership=pit_membership,
min_price=5.0,
min_dollar_volume=1_000.0,
min_history_days=1,
min_valid_volume_days=1,
liquidity_window=1,
)
expected = pd.DataFrame({"AAA": [False, False, False]}, index=dates, dtype=bool)
pd.testing.assert_frame_equal(mask, expected)
def test_build_tradable_mask_uses_strict_thresholds(self):
from research.us_universe import build_tradable_mask
dates = pd.date_range("2024-01-01", periods=3, freq="D")
close = pd.DataFrame({"AAA": [5.0, 5.0, 5.0]}, index=dates)
volume = pd.DataFrame({"AAA": [300.0, 300.0, 300.0]}, index=dates)
mask = build_tradable_mask(
close=close,
volume=volume,
pit_membership=None,
min_price=5.0,
min_dollar_volume=1_000.0,
min_history_days=1,
min_valid_volume_days=1,
liquidity_window=1,
)
expected = pd.DataFrame({"AAA": [False, False, False]}, index=dates, dtype=bool)
pd.testing.assert_frame_equal(mask, expected)
def test_build_tradable_mask_uses_strict_dollar_volume_threshold(self):
from research.us_universe import build_tradable_mask
dates = pd.date_range("2024-01-01", periods=3, freq="D")
close = pd.DataFrame({"AAA": [8.0, 8.0, 8.0]}, index=dates)
volume = pd.DataFrame({"AAA": [125.0, 125.0, 125.0]}, index=dates)
mask = build_tradable_mask(
close=close,
volume=volume,
pit_membership=None,
min_price=5.0,
min_dollar_volume=1_000.0,
min_history_days=1,
min_valid_volume_days=1,
liquidity_window=1,
)
expected = pd.DataFrame({"AAA": [False, False, False]}, index=dates, dtype=bool)
pd.testing.assert_frame_equal(mask, expected)
def test_build_tradable_mask_requires_valid_dollar_volume_history(self):
from research.us_universe import build_tradable_mask
dates = pd.date_range("2024-01-01", periods=4, freq="D")
close = pd.DataFrame({"AAA": [10.0, float("nan"), 10.0, 10.0]}, index=dates)
volume = pd.DataFrame({"AAA": [200.0, 200.0, 200.0, 200.0]}, index=dates)
mask = build_tradable_mask(
close=close,
volume=volume,
pit_membership=None,
min_price=5.0,
min_dollar_volume=1_000.0,
min_history_days=1,
min_valid_volume_days=2,
liquidity_window=2,
)
expected = pd.DataFrame({"AAA": [False, False, False, False]}, index=dates, dtype=bool)
pd.testing.assert_frame_equal(mask, expected)
if __name__ == "__main__":
unittest.main()