diff --git a/setup.cfg b/setup.cfg index 4edd93f..4e57090 100644 --- a/setup.cfg +++ b/setup.cfg @@ -12,7 +12,7 @@ max-line-length = 120 ignore = E501, W503, E202, E226 [tool:pytest] -#addopts = --cov=. +addopts = --cov=. markers = only_on_pr: marks tests as slow (select with -m only_on_pr and deselect with -m "not only_on_pr") pythonpath = ./tests diff --git a/tests/tools/test_hedge1.py b/tests/tools/test_hedge1.py new file mode 100644 index 0000000..e3d1008 --- /dev/null +++ b/tests/tools/test_hedge1.py @@ -0,0 +1,105 @@ +"""Testing hedge functions using hard-coded hedge values.""" + +import numpy as np +import pandas as pd +import pytest + +from portfolyo import testing, tools + + +@pytest.mark.parametrize("how", ["vol", "val"]) +@pytest.mark.parametrize( + "w_vals,start,w_expected", + [ + ([1, 2, 3], "2020-01-01", 2), + (range(12), "2020-01-01", 5.5), + ([*[10] * (23 + 8), *[15] * 5], "2020-03-29", 10.69444), + ([*[10] * (25 + 8), *[15] * 5], "2020-10-25", 10.65789), + ], +) +def test_onehedge_uniformpricesandduration(w_vals, start, w_expected, how): + """Test hedge with uniform prices and durations.""" + i = pd.date_range(start, freq="H", periods=len(w_vals), tz="Europe/Berlin") + df = pd.DataFrame({"w": w_vals, "p": 100.0}, i) + df["duration"] = tools.duration.index(i) + + result = tools.hedge.one_hedge(df, how=how) + expected = pd.Series({"w": w_expected, "p": 100.0}) + testing.assert_series_equal(result, expected) + + +@pytest.mark.parametrize("how", ["vol", "val"]) +@pytest.mark.parametrize( + "w_vals,start,w_expected,tz", + [ + ([1, 2], "2020-01-01", 1.483333333, None), # 29 days in Feb + ([1, 2], "2021-01-01", 1.474576271, None), # 28 days in Feb + (range(12), "2020-01-01", 5.51366120, None), # no DST + (range(12), "2020-01-01", 5.514458106, "Europe/Berlin"), # DST + ], +) +def test_onehedge_uniformprices(w_vals, start, w_expected, how, tz): + """Test hedge with uniform prices but distinct durations.""" + i = pd.date_range(start, freq="MS", periods=len(w_vals), tz=tz) + df = pd.DataFrame({"w": w_vals, "p": 100.0}, i) + df["duration"] = tools.duration.index(i) + + result = tools.hedge.one_hedge(df, how=how) + expected = pd.Series({"w": w_expected, "p": 100.0}) + testing.assert_series_equal(result, expected) + + +@pytest.mark.parametrize("how", ["vol", "val"]) +@pytest.mark.parametrize( + "w_vals,p_vals,start,p_expected,w_expected_vol,w_expected_val,tz", + [ + ( + [1, 2], + [100, 200], + "2020-01-01", + 148.333333333, + 1.4833333, + 1.651685393, + None, + ), # 29 days in Feb + ( + [1, 2], + [100, 200], + "2021-01-01", + 147.4576271, + 1.474576271, + 1.643678161, + None, + ), # 28 days in Feb + ( + np.arange(12), + np.arange(12) * 100, + "2020-01-01", + 551.366120, + 5.51366120, + 7.673934589, + None, + ), # no DST + ( + np.arange(12), + np.arange(12) * 100, + "2020-01-01", + 551.4458106, + 5.514458106, + 7.674415244, + "Europe/Berlin", + ), # DST + ], +) +def test_onehedge( + w_vals, p_vals, start, w_expected_val, w_expected_vol, p_expected, how, tz +): + """Test value hedge.""" + i = pd.date_range(start, freq="MS", periods=len(w_vals), tz=tz) + df = pd.DataFrame({"w": w_vals, "p": p_vals}, i) + df["duration"] = tools.duration.index(i) + + result = tools.hedge.one_hedge(df, how=how) + w_expected = w_expected_val if how == "val" else w_expected_vol + expected = pd.Series({"w": w_expected, "p": p_expected}) + testing.assert_series_equal(result, expected) diff --git a/tests/tools/test_hedge2.py b/tests/tools/test_hedge2.py new file mode 100644 index 0000000..bc82b59 --- /dev/null +++ b/tests/tools/test_hedge2.py @@ -0,0 +1,127 @@ +"""Testing hedge functions using pre-calculated hedge values from excel.""" + +from pathlib import Path + +import pandas as pd +import pytest + +from portfolyo import testing, tools + +PATH = Path(__file__).parent / "test_hedge_data.xlsx" + + +def sheetname(freq, tz): + return f'{freq}_{"None" if tz is None else tz.replace("/", "")}' + + +@pytest.fixture( + scope="session", + params=[pytest.param(True, id="withunits"), pytest.param(False, id="withoutunits")], +) +def has_units(request): + return request.param + + +@pytest.fixture(scope="session", params=["H", "D"]) +def freq(request): + return request.param + + +@pytest.fixture(scope="session", params=[None, "Europe/Berlin"]) +def tz(request): + return request.param + + +@pytest.fixture(scope="session", params=["vol", "val"]) +def how(request): + return request.param + + +@pytest.fixture( + scope="session", + params=[ + pytest.param(None, id="base"), + pytest.param(tools.product.germanpower_peakfn, id="peakoffpeakgerman"), + ], +) +def peak_fn(request): + return request.param + + +@pytest.fixture(scope="session", params=["MS", "QS", "AS"]) +def aggfreq(request): + return request.param + + +@pytest.fixture +def dfin(freq: str, tz: str) -> pd.DataFrame: + df = pd.read_excel( + PATH, sheetname(freq, tz), header=6, index_col=0, usecols="A,B:C" + ) + if tz: + df = df.tz_localize(tz, ambiguous="infer") + df.index.freq = pd.infer_freq(df.index) + return df + + +@pytest.fixture +def dfexp(freq: str, tz: str) -> pd.DataFrame: + df = pd.read_excel( + PATH, f"{sheetname(freq, tz)}_out", header=[3, 4, 5, 6], index_col=0 + ) + if tz: + df = df.tz_localize(tz, ambiguous="infer") + df.index.freq = pd.infer_freq(df.index) + return df + + +@pytest.fixture +def win(dfin: pd.DataFrame, has_units: bool) -> pd.Series: + s = dfin["w"] + return s if not has_units else s.astype("pint[MW]") + + +@pytest.fixture +def pin(dfin: pd.DataFrame, has_units: bool) -> pd.Series: + s = dfin["p"] + return s if not has_units else s.astype("pint[Eur/MWh]") + + +@pytest.fixture +def wexp( + dfexp: pd.DataFrame, + has_units: bool, + aggfreq: str, + peak_fn: tools.peakfn.PeakFunction, + how: str, +) -> pd.Series: + if dfexp.index.freq == "D" and peak_fn is not None: + pytest.skip("Don't decompose in peak and offpeak if daily values") + s = dfexp[aggfreq][peak_fn is not None][how]["w"] + return s if not has_units else s.astype("pint[MW]") + + +@pytest.fixture +def pexp( + dfexp: pd.DataFrame, + has_units: bool, + aggfreq: str, + peak_fn: tools.peakfn.PeakFunction, +) -> pd.Series: + s = dfexp[aggfreq][peak_fn is not None].iloc[:, 0] # price is first column + return s if not has_units else s.astype("pint[Eur/MWh]") + + +def test_hedge_fromexcel( + win: pd.Series, + pin: pd.Series, + wexp: pd.Series, + pexp: pd.Series, + how: str, + aggfreq: str, + peak_fn: tools.peakfn.PeakFunction, +): + """Test if hedge results are correctly calculated, by comparing against previously calculated results.""" + wresult, presult = tools.hedge.hedge(win, pin, how, peak_fn, aggfreq) + testing.assert_series_equal(wresult, wexp, check_names=False) + testing.assert_series_equal(presult, pexp, check_names=False)