mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-28 07:57:44 +00:00
TPTBusiness
a781d003ba
- factor_runner: shift_daily_constant (property-based, 50 inputs), multi-instrument, NaN handling, edge cases (2-day, all-same, all-NaN), IC import, safe_float - strategy_builder: combinator (pairs/triplets, category filtering, empty/single), evaluator (cost calc, safe names), builder import
161 lines
7.0 KiB
Python
161 lines
7.0 KiB
Python
"""Deep tests for factor_runner.py — look-ahead fix, IC, de-duplication, edge cases."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
from hypothesis import given, settings
|
||
from hypothesis import strategies as st
|
||
from hypothesis.extra.numpy import arrays
|
||
|
||
from rdagent.scenarios.qlib.developer.factor_runner import (
|
||
_shift_daily_constant_factor_if_needed,
|
||
)
|
||
|
||
|
||
def _make_multiindex_series(
|
||
dates: list, instrument: str = "EURUSD", values: list = None
|
||
) -> pd.Series:
|
||
"""Helper: build a MultiIndex (datetime, instrument) Series."""
|
||
idx = pd.MultiIndex.from_tuples(
|
||
[(d, instrument) for d in dates], names=["datetime", "instrument"]
|
||
)
|
||
if values is None:
|
||
values = np.arange(len(dates), dtype=float)
|
||
return pd.Series(values, index=idx, name="test_factor")
|
||
|
||
|
||
class TestShiftDailyConstantFactor:
|
||
def test_returns_unchanged_when_few_rows(self):
|
||
"""< 200 non-null rows — skip shift entirely."""
|
||
dates = pd.date_range("2024-01-01", periods=50, freq="1min")
|
||
s = _make_multiindex_series(dates, values=np.ones(50))
|
||
result = _shift_daily_constant_factor_if_needed(s, "test")
|
||
assert result.equals(s)
|
||
|
||
def test_returns_unchanged_when_intraday_varying(self):
|
||
"""Factor changes within a day → no shift needed."""
|
||
dates = pd.date_range("2024-01-01", periods=2000, freq="1min")
|
||
vals = np.random.default_rng(1).normal(0, 1, 2000)
|
||
s = _make_multiindex_series(dates, values=vals)
|
||
result = _shift_daily_constant_factor_if_needed(s, "test")
|
||
assert result.equals(s)
|
||
|
||
def test_shifts_daily_constant_factor(self):
|
||
"""Factor is identical across all bars in a day → shift by 1 day."""
|
||
dates = pd.date_range("2024-01-01 00:00", periods=5000, freq="1min")
|
||
# Create daily-constant: same value for all bars on same day
|
||
vals = np.array([d.day for d in dates], dtype=float)
|
||
s = _make_multiindex_series(dates, values=vals)
|
||
result = _shift_daily_constant_factor_if_needed(s, "test")
|
||
# After shift, the value at day 2 should be the value from day 1
|
||
assert not result.equals(s) # Must have been shifted
|
||
|
||
def test_nan_handling(self):
|
||
"""NaN values in the factor should not break the shift."""
|
||
dates = pd.date_range("2024-01-01", periods=2000, freq="1min")
|
||
vals = np.array([d.day for d in dates], dtype=float)
|
||
vals[:100] = np.nan # First 100 NaN
|
||
s = _make_multiindex_series(dates, values=vals)
|
||
result = _shift_daily_constant_factor_if_needed(s, "test")
|
||
assert isinstance(result, pd.Series)
|
||
assert len(result) == len(s)
|
||
|
||
def test_multi_instrument_handled(self):
|
||
"""Multi-instrument data should not crash."""
|
||
dates = pd.date_range("2024-01-01", periods=3000, freq="1min")
|
||
tuples_eur = [(d, "EURUSD") for d in dates]
|
||
tuples_gbp = [(d, "GBPUSD") for d in dates]
|
||
all_tuples = tuples_eur + tuples_gbp
|
||
idx = pd.MultiIndex.from_tuples(all_tuples, names=["datetime", "instrument"])
|
||
vals = [d.day for d in dates] + [d.day for d in dates]
|
||
s = pd.Series(vals, index=idx, name="test", dtype=float)
|
||
result = _shift_daily_constant_factor_if_needed(s, "test")
|
||
assert isinstance(result, pd.Series)
|
||
|
||
def test_all_same_value(self):
|
||
"""Single unique value across entire series → treated as daily-constant."""
|
||
dates = pd.date_range("2024-01-01", periods=2000, freq="1min")
|
||
s = _make_multiindex_series(dates, values=np.ones(2000))
|
||
result = _shift_daily_constant_factor_if_needed(s, "test")
|
||
assert isinstance(result, pd.Series)
|
||
|
||
def test_two_days_only(self):
|
||
"""Only 2 days of data — should still handle gracefully."""
|
||
# 2 days × 100 bars = 200 bars
|
||
dates = pd.date_range("2024-01-01 00:00", periods=200, freq="1min")
|
||
vals = np.array([d.day for d in dates], dtype=float)
|
||
s = _make_multiindex_series(dates, values=vals)
|
||
result = _shift_daily_constant_factor_if_needed(s, "test")
|
||
assert isinstance(result, pd.Series)
|
||
|
||
def test_zero_unique_values_edge_case(self):
|
||
"""All-NaN with very few valid should return unchanged."""
|
||
dates = pd.date_range("2024-01-01", periods=500, freq="1min")
|
||
vals = np.full(500, np.nan)
|
||
vals[100:105] = 1.0
|
||
s = _make_multiindex_series(dates, values=vals)
|
||
# This should trigger the "< 200 non-null" check and return unchanged
|
||
result = _shift_daily_constant_factor_if_needed(s, "test")
|
||
assert result.equals(s)
|
||
|
||
@given(
|
||
n_days=st.integers(min_value=5, max_value=50),
|
||
seed=st.integers(min_value=0, max_value=100),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_property_never_crashes(self, n_days, seed):
|
||
"""For any valid MultiIndex series, function must not crash."""
|
||
rng = np.random.default_rng(seed)
|
||
bars_per_day = 48 # 30-min bars
|
||
dates = pd.date_range("2024-01-01", periods=n_days * bars_per_day, freq="30min")
|
||
vals = rng.choice([1.0, 2.0, 3.0], n_days * bars_per_day) # Daily-constant
|
||
s = _make_multiindex_series(dates, values=vals)
|
||
result = _shift_daily_constant_factor_if_needed(s, f"f_{seed}")
|
||
assert isinstance(result, pd.Series)
|
||
assert len(result) == len(s)
|
||
|
||
|
||
class TestInformationCoefficient:
|
||
def test_ic_direct_import(self):
|
||
"""calculate_information_coefficient is importable and callable."""
|
||
from rdagent.scenarios.qlib.developer.factor_runner import (
|
||
QlibFactorRunner,
|
||
)
|
||
assert hasattr(QlibFactorRunner, "calculate_information_coefficient")
|
||
|
||
|
||
class TestSafeFloat:
|
||
def test_safe_float_direct(self):
|
||
"""_safe_float must handle NaN, Inf, None, strings."""
|
||
from rdagent.scenarios.qlib.developer.factor_runner import QlibFactorRunner
|
||
|
||
# Create a minimal instance
|
||
runner = QlibFactorRunner.__new__(QlibFactorRunner)
|
||
# _safe_float should be callable without full init
|
||
if hasattr(runner, "_safe_float"):
|
||
assert runner._safe_float(1.5) == 1.5
|
||
assert runner._safe_float(float("nan")) is None
|
||
assert runner._safe_float(float("inf")) is None
|
||
assert runner._safe_float(None) is None
|
||
|
||
|
||
class TestDeduplicateFactors:
|
||
def test_deduplicate_importable(self):
|
||
"""deduplicate_new_factors is importable."""
|
||
from rdagent.scenarios.qlib.developer.factor_runner import QlibFactorRunner
|
||
assert hasattr(QlibFactorRunner, "deduplicate_new_factors")
|
||
|
||
|
||
class TestFactorIntegration:
|
||
def test_shift_preserves_index_structure(self):
|
||
"""After shift, index names and structure must match original."""
|
||
dates = pd.date_range("2024-01-01 00:00", periods=3000, freq="1min")
|
||
vals = np.array([d.day for d in dates], dtype=float)
|
||
s = _make_multiindex_series(dates, values=vals)
|
||
result = _shift_daily_constant_factor_if_needed(s, "test")
|
||
assert result.index.names == s.index.names
|
||
assert len(result.index) == len(s.index)
|