Coverage for src/stable_yield_lab/sources/base.py: 89%

1"""Base utilities for StableYieldLab data sources.""" 

2 

3from __future__ import annotations 

4 

5import pandas as pd 

6 

7from ..core import PoolReturn 

8 

9 

10class HistoricalCSVSource: 

11 """Load periodic returns from a CSV with timestamp, name and period_return.""" 

12 

13 def __init__(self, path: str) -> None: 

14 self.path = path 

15 

16 def fetch(self) -> list[PoolReturn]: 

17 df = pd.read_csv(self.path) 

18 required = {"timestamp", "name", "period_return"} 

19 missing = required.difference(df.columns) 

20 if missing: 20 ↛ 21line 20 didn't jump to line 21 because the condition on line 20 was never true

21 raise ValueError(f"CSV missing columns: {missing}") 

22 df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True) 

23 rows = [ 

24 PoolReturn( 

25 name=str(r["name"]), 

26 timestamp=pd.Timestamp(r["timestamp"]), 

27 period_return=float(r["period_return"]), 

28 ) 

29 for _, r in df.iterrows() 

30 ] 

31 return rows 

32 

33 

34__all__ = ["HistoricalCSVSource"]