diff --git a/src/quant_platform_kit/common/history.py b/src/quant_platform_kit/common/history.py new file mode 100644 index 0000000..9e992a0 --- /dev/null +++ b/src/quant_platform_kit/common/history.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +import pandas as pd + + +def normalize_history_frame(history, *, label: str, close_column: str = "close") -> pd.DataFrame: + if isinstance(history, pd.DataFrame): + frame = history.copy() + elif isinstance(history, pd.Series): + frame = history.to_frame(name=close_column) + else: + frame = pd.DataFrame(list(history)) + + if frame.empty: + raise ValueError(f"{label} must contain close history") + + normalized_close = str(close_column or "close").strip() or "close" + lower_columns = {str(column).strip().lower(): column for column in frame.columns} + if normalized_close not in frame.columns: + close_match = lower_columns.get(normalized_close.lower()) + if close_match is not None: + frame = frame.rename(columns={close_match: normalized_close}) + elif len(frame.columns) == 1: + frame = frame.rename(columns={frame.columns[0]: normalized_close}) + else: + columns = ", ".join(str(column) for column in frame.columns) + raise ValueError(f"{label} must include a {normalized_close} column; got columns: {columns or ''}") + + frame[normalized_close] = pd.to_numeric(frame[normalized_close], errors="coerce") + frame = frame.dropna(subset=[normalized_close]).reset_index(drop=True) + if frame.empty: + raise ValueError(f"{label} close history is empty after normalization") + return frame diff --git a/tests/test_common_history.py b/tests/test_common_history.py new file mode 100644 index 0000000..9f1b864 --- /dev/null +++ b/tests/test_common_history.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import unittest + +import pandas as pd + +from quant_platform_kit.common.history import normalize_history_frame + + +class CommonHistoryTests(unittest.TestCase): + def test_normalize_history_frame_accepts_dataframe_with_close_column(self) -> None: + frame = pd.DataFrame([{"close": 1.0}, {"close": 2.0}]) + + normalized = normalize_history_frame(frame, label="benchmark_history") + + self.assertEqual(list(normalized["close"]), [1.0, 2.0]) + + def test_normalize_history_frame_accepts_dataframe_with_close_case_variant(self) -> None: + frame = pd.DataFrame([{"Close": 1.0}, {"Close": 2.0}]) + + normalized = normalize_history_frame(frame, label="benchmark_history") + + self.assertEqual(list(normalized["close"]), [1.0, 2.0]) + + def test_normalize_history_frame_accepts_series(self) -> None: + series = pd.Series([1.0, 2.0, 3.0], name="close") + + normalized = normalize_history_frame(series, label="benchmark_history") + + self.assertEqual(list(normalized["close"]), [1.0, 2.0, 3.0]) + + def test_normalize_history_frame_accepts_single_column_iterable(self) -> None: + history = [1.0, 2.0, 3.0] + + normalized = normalize_history_frame(history, label="benchmark_history") + + self.assertEqual(list(normalized["close"]), [1.0, 2.0, 3.0]) + + def test_normalize_history_frame_rejects_missing_close_data(self) -> None: + with self.assertRaisesRegex(ValueError, "benchmark_history must include a close column"): + normalize_history_frame([{"open": 1.0, "high": 1.0}], label="benchmark_history") + + +if __name__ == "__main__": + unittest.main()