|
4 | 4 |
|
5 | 5 | from quanteval.strategies.multi_factor_model import MultiFactorModel |
6 | 6 | from quanteval.aggregators.weighted import ( |
7 | | - EqualWeightAggregator, |
8 | | - ICWeightedAggregator, |
| 7 | + EqualWeightAggregator, |
| 8 | + ICWeightedAggregator, |
9 | 9 | ) |
10 | 10 | from quanteval.factors.base import Factor |
11 | 11 |
|
12 | 12 |
|
13 | 13 | def test_equal_weight_aggregator_without_normalize(sample_market_data: pd.DataFrame) -> None: |
14 | | - data = sample_market_data |
15 | | - idx = data.index |
16 | | - s1 = data['Close'].pct_change().fillna(0) |
17 | | - s2 = -s1 |
| 14 | + data = sample_market_data |
| 15 | + idx = data.index |
| 16 | + s1 = data['Close'].pct_change().fillna(0) |
| 17 | + s2 = -s1 |
18 | 18 |
|
19 | | - agg = EqualWeightAggregator(normalize=False) |
20 | | - composite = agg.aggregate({'s1': s1, 's2': s2}, data) |
| 19 | + agg = EqualWeightAggregator(normalize=False) |
| 20 | + composite = agg.aggregate({'s1': s1, 's2': s2}, data) |
21 | 21 |
|
22 | | - expected = (s1.reindex(idx) + s2.reindex(idx)) / 2 |
23 | | - pd.testing.assert_series_equal(composite, expected, check_names=False) |
| 22 | + expected = (s1.reindex(idx) + s2.reindex(idx)) / 2 |
| 23 | + pd.testing.assert_series_equal(composite, expected, check_names=False) |
24 | 24 |
|
25 | 25 |
|
26 | 26 | def test_icweighted_fallback_to_equal_weights(sample_market_data: pd.DataFrame) -> None: |
27 | | - # Force forward returns to be NaN so IC is unavailable and aggregator falls |
28 | | - # back to equal weights. |
29 | | - data = sample_market_data.copy() |
30 | | - data['Close'] = np.nan |
31 | | - idx = data.index |
| 27 | + # Force forward returns to be NaN so IC is unavailable and aggregator falls |
| 28 | + # back to equal weights. |
| 29 | + data = sample_market_data.copy() |
| 30 | + data['Close'] = np.nan |
| 31 | + idx = data.index |
32 | 32 |
|
33 | | - s1 = pd.Series(np.arange(len(idx)), index=idx, name='s1') |
34 | | - s2 = pd.Series(np.arange(len(idx)) * 2.0, index=idx, name='s2') |
| 33 | + s1 = pd.Series(np.arange(len(idx)), index=idx, name='s1') |
| 34 | + s2 = pd.Series(np.arange(len(idx)) * 2.0, index=idx, name='s2') |
35 | 35 |
|
36 | | - agg = ICWeightedAggregator(normalize=False, ic_window=3, min_periods=1) |
37 | | - composite = agg.aggregate({'s1': s1, 's2': s2}, data) |
| 36 | + agg = ICWeightedAggregator(normalize=False, ic_window=3, min_periods=1) |
| 37 | + composite = agg.aggregate({'s1': s1, 's2': s2}, data) |
38 | 38 |
|
39 | | - expected = (s1 + s2) / 2 |
40 | | - pd.testing.assert_series_equal(composite, expected, check_names=False) |
| 39 | + expected = (s1 + s2) / 2 |
| 40 | + pd.testing.assert_series_equal(composite, expected, check_names=False) |
41 | 41 |
|
42 | 42 |
|
43 | 43 | class ConstantFactor(Factor): |
44 | | - def __init__(self, name: str, series: pd.Series): |
45 | | - super().__init__(name=name) |
46 | | - self._series = series |
| 44 | + def __init__(self, name: str, series: pd.Series): |
| 45 | + super().__init__(name=name) |
| 46 | + self._series = series |
47 | 47 |
|
48 | | - def calculate(self, data: pd.DataFrame) -> pd.Series: |
49 | | - return self._series.reindex(data.index) |
| 48 | + def calculate(self, data: pd.DataFrame) -> pd.Series: |
| 49 | + return self._series.reindex(data.index) |
50 | 50 |
|
51 | 51 |
|
52 | 52 | def test_multi_factor_model_signals_and_type_error(sample_market_data: pd.DataFrame) -> None: |
53 | | - data = sample_market_data.iloc[:4] |
54 | | - idx = data.index |
| 53 | + data = sample_market_data.iloc[:4] |
| 54 | + idx = data.index |
55 | 55 |
|
56 | | - s = pd.Series([0.0, 1.0, 0.0, 1.0], index=idx) |
57 | | - f1 = ConstantFactor('f1', s) |
58 | | - f2 = ConstantFactor('f2', s) |
| 56 | + s = pd.Series([0.0, 1.0, 0.0, 1.0], index=idx) |
| 57 | + f1 = ConstantFactor('f1', s) |
| 58 | + f2 = ConstantFactor('f2', s) |
59 | 59 |
|
60 | | - agg = EqualWeightAggregator(normalize=False) |
61 | | - model = MultiFactorModel([f1, f2], aggregator=agg, threshold=0.5) |
| 60 | + agg = EqualWeightAggregator(normalize=False) |
| 61 | + model = MultiFactorModel([f1, f2], aggregator=agg, threshold=0.5) |
62 | 62 |
|
63 | | - signals = model.generate_signals(data) |
64 | | - expected = (((s + s) / 2) > 0.5).astype(float).ffill().fillna(0.0) |
65 | | - expected.name = signals.name |
66 | | - pd.testing.assert_series_equal(signals, expected) |
| 63 | + signals = model.generate_signals(data) |
| 64 | + expected = (((s + s) / 2) > 0.5).astype(float).ffill().fillna(0.0) |
| 65 | + expected.name = signals.name |
| 66 | + pd.testing.assert_series_equal(signals, expected) |
67 | 67 |
|
68 | | - # Factor returning a dict should raise TypeError |
69 | | - class DictFactor(Factor): |
70 | | - def __init__(self): |
71 | | - super().__init__(name='dict') |
| 68 | + # Factor returning a dict should raise TypeError |
| 69 | + class DictFactor(Factor): |
| 70 | + def __init__(self): |
| 71 | + super().__init__(name='dict') |
72 | 72 |
|
73 | | - def calculate(self, data: pd.DataFrame): |
74 | | - return {'a': pd.Series([1] * len(data), index=data.index)} |
75 | | - |
76 | | - bad = DictFactor() |
77 | | - model2 = MultiFactorModel([bad], aggregator=agg) |
78 | | - with pytest.raises(TypeError): |
79 | | - model2.generate_signals(data) |
| 73 | + def calculate(self, data: pd.DataFrame): |
| 74 | + return {'a': pd.Series([1] * len(data), index=data.index)} |
80 | 75 |
|
| 76 | + bad = DictFactor() |
| 77 | + model2 = MultiFactorModel([bad], aggregator=agg) |
| 78 | + with pytest.raises(TypeError): |
| 79 | + model2.generate_signals(data) |
0 commit comments