diff --git a/src/argus/analytics/metrics/trend_metrics.py b/src/argus/analytics/metrics/trend_metrics.py index 5622acb..40e857c 100644 --- a/src/argus/analytics/metrics/trend_metrics.py +++ b/src/argus/analytics/metrics/trend_metrics.py @@ -73,3 +73,57 @@ def get_min_max_rates(df: pd.DataFrame) -> dict: min_max["max_date"].append(df.loc[max_id, "date"]) min_max["max_rate"].append(df.loc[max_id, "rate"]) return min_max + + +def get_cumulative_return(df: pd.DataFrame) -> float: + if df.empty: + return 0.0 + + start_rate = float(df["rate"].iloc[0]) + end_rate = float(df["rate"].iloc[-1]) + + if start_rate == 0.0: + return 0.0 + + return (end_rate - start_rate) / start_rate * 100 + + +def get_strongest_weakest_days(df: pd.DataFrame) -> dict: + if df.empty or len(df) < 2: + return { + "strongest_day": {"date": None, "pct_change": 0.0}, + "weakest_day": {"date": None, "pct_change": 0.0}, + } + + pct_series = df.loc[:, "rate"].pct_change() * 100 + valid_pct = pct_series.dropna() + + if valid_pct.empty: + return { + "strongest_day": {"date": None, "pct_change": 0.0}, + "weakest_day": {"date": None, "pct_change": 0.0}, + } + + max_idx = valid_pct.idxmax() + min_idx = valid_pct.idxmin() + + return { + "strongest_day": { + "date": df.loc[max_idx, "date"], + "pct_change": round(float(pct_series.loc[max_idx]), 2), + }, + "weakest_day": { + "date": df.loc[min_idx, "date"], + "pct_change": round(float(pct_series.loc[min_idx]), 2), + }, + } + + +def add_rolling_volatility(df: pd.DataFrame, window: int = 3) -> pd.DataFrame: + result = df.copy() + daily_returns = result["rate"].pct_change() * 100 + result["rolling_volatility"] = daily_returns.rolling( + window=window, min_periods=1 + ).std() + result["rolling_volatility"] = result["rolling_volatility"].fillna(0.0) + return result diff --git a/tests/test_trend_metrics.py b/tests/test_trend_metrics.py index fe175bd..5a3caac 100644 --- a/tests/test_trend_metrics.py +++ b/tests/test_trend_metrics.py @@ -1,10 +1,14 @@ import pandas as pd import pandas.testing as pdt import numpy as np +import pytest from argus.analytics.metrics.trend_metrics import ( add_daily_percentage_change, add_rolling_average, get_min_max_rates, + get_cumulative_return, + get_strongest_weakest_days, + add_rolling_volatility, ) @@ -60,3 +64,63 @@ def test_get_min_max_(): result_dict = get_min_max_rates(test_df) assert result_dict == min_max + + +def test_get_cumulative_return(): + test_timesseries = { + "date": ["2026-05-01", "2026-05-02", "2026-05-03"], + "rate": [1.00, 1.10, 1.21], + } + test_df = pd.DataFrame(test_timesseries) + resault = get_cumulative_return(test_df) + assert resault == pytest.approx(21.0) + + # Egde case + empty_df = pd.DataFrame(columns=["date", "rate"]) + result = get_cumulative_return(empty_df) + assert result == 0.0 + + +def test_get_strongest_weakest_days(): + test_timeseries = { + "date": ["2026-05-01", "2026-05-02", "2026-05-03", "2026-05-04"], + "rate": [1.00, 1.20, 1.14, 2.00], + } + test_df = pd.DataFrame(test_timeseries) + + result = get_strongest_weakest_days(test_df) + + assert result == { + "strongest_day": {"date": "2026-05-04", "pct_change": 75.44}, + "weakest_day": {"date": "2026-05-03", "pct_change": -5.0}, + } + + # Edge case + flat_timeseries = { + "date": ["2026-05-01", "2026-05-02", "2026-05-03"], + "rate": [1.15, 1.15, 1.15], + } + flat_df = pd.DataFrame(flat_timeseries) + result = get_strongest_weakest_days(flat_df) + assert result == { + "strongest_day": {"date": "2026-05-02", "pct_change": 0.0}, + "weakest_day": {"date": "2026-05-02", "pct_change": 0.0}, + } + + +def test_is_rolling_volatility_added(): + test_timeseries = { + "date": ["2026-05-01", "2026-05-02", "2026-05-03"], + "rate": [1.00, 2.00, 1.00], + } + test_df = pd.DataFrame(test_timeseries) + + expect_result = { + "date": ["2026-05-01", "2026-05-02", "2026-05-03"], + "rate": [1.00, 2.00, 1.00], + "rolling_volatility": [0.0, 0.0, 106.06601717798213], + } + expect_df = pd.DataFrame(expect_result) + result_df = add_rolling_volatility(test_df, window=2) + + pdt.assert_frame_equal(result_df, expect_df)