diff --git a/src/ml4t/data/macro/downloader.py b/src/ml4t/data/macro/downloader.py index 115755a1..18a85158 100644 --- a/src/ml4t/data/macro/downloader.py +++ b/src/ml4t/data/macro/downloader.py @@ -30,6 +30,7 @@ import os from dataclasses import dataclass, field +from datetime import datetime, timedelta from pathlib import Path from typing import Any @@ -269,10 +270,18 @@ def _download_from_yfinance(self, symbols: list[str]) -> pl.DataFrame: ticker = yf_mapping[series_id] try: + # yfinance treats end as exclusive. Config dates are inclusive. + end = self.config.end + try: + end = (datetime.strptime(end, "%Y-%m-%d") + timedelta(days=1)).strftime( + "%Y-%m-%d" + ) + except ValueError: + pass df = yf.download( ticker, start=self.config.start, - end=self.config.end, + end=end, progress=False, auto_adjust=True, threads=False, diff --git a/src/ml4t/data/providers/fama_french.py b/src/ml4t/data/providers/fama_french.py index bc899d08..3b618a4d 100644 --- a/src/ml4t/data/providers/fama_french.py +++ b/src/ml4t/data/providers/fama_french.py @@ -41,6 +41,7 @@ import io import zipfile +from calendar import monthrange from datetime import datetime from pathlib import Path from typing import Any, ClassVar, Literal @@ -730,11 +731,11 @@ def fetch( ) df = df.filter(pl.col("timestamp") >= start_date) if end: - end_date = ( - datetime.strptime(end[:10], "%Y-%m-%d").date() - if len(end) >= 10 - else datetime.strptime(end + "-28", "%Y-%m-%d").date() - ) + if len(end) >= 10: + end_date = datetime.strptime(end[:10], "%Y-%m-%d").date() + else: + year, month = (int(part) for part in end.split("-", 1)) + end_date = datetime(year, month, monthrange(year, month)[1]).date() df = df.filter(pl.col("timestamp") <= end_date) self.logger.info( diff --git a/src/ml4t/data/providers/finnhub.py b/src/ml4t/data/providers/finnhub.py index e80ff1e9..3dca8715 100644 --- a/src/ml4t/data/providers/finnhub.py +++ b/src/ml4t/data/providers/finnhub.py @@ -4,7 +4,7 @@ """ import os -from datetime import datetime +from datetime import UTC, datetime from typing import Any, ClassVar import polars as pl @@ -34,6 +34,19 @@ logger = structlog.get_logger() +def _utc_unix_timestamp(value: str, *, inclusive_date: bool = False) -> int: + """Parse a date or datetime as a UTC unix timestamp. + + A date-only end bound covers that UTC calendar day. An explicit datetime + is used as given, with naive values treated as UTC. + """ + parsed = datetime.fromisoformat(value) + parsed = parsed.replace(tzinfo=UTC) if parsed.tzinfo is None else parsed.astimezone(UTC) + if inclusive_date and "T" not in value and len(value) == 10: + parsed = parsed.replace(hour=23, minute=59, second=59) + return int(parsed.timestamp()) + + class FinnhubProvider(BaseProvider): """Fetch Finnhub market data through synchronous provider methods. @@ -196,12 +209,11 @@ def _fetch_raw_data( value=frequency, ) - # Convert dates to unix timestamps + # Convert dates to unix timestamps. Date-only bounds are UTC calendar + # days, with the end date inclusive through 23:59:59. try: - start_dt = datetime.fromisoformat(start) - end_dt = datetime.fromisoformat(end) - start_ts = int(start_dt.timestamp()) - end_ts = int(end_dt.timestamp()) + start_ts = _utc_unix_timestamp(start) + end_ts = _utc_unix_timestamp(end, inclusive_date=True) except ValueError as err: raise DataValidationError( provider="finnhub", diff --git a/src/ml4t/data/providers/kalshi.py b/src/ml4t/data/providers/kalshi.py index 2385de89..9a415a57 100644 --- a/src/ml4t/data/providers/kalshi.py +++ b/src/ml4t/data/providers/kalshi.py @@ -35,7 +35,7 @@ """ import time -from datetime import datetime +from datetime import UTC, datetime from typing import Any, ClassVar import polars as pl @@ -201,11 +201,11 @@ def _fetch_raw_data( value=frequency, ) - # Convert dates to unix timestamps - start_dt = datetime.strptime(start, "%Y-%m-%d") - end_dt = datetime.strptime(end, "%Y-%m-%d") - # Set end to end of day - end_dt = end_dt.replace(hour=23, minute=59, second=59) + # Convert inclusive UTC calendar dates to unix timestamps. + start_dt = datetime.strptime(start, "%Y-%m-%d").replace(tzinfo=UTC) + end_dt = datetime.strptime(end, "%Y-%m-%d").replace( + hour=23, minute=59, second=59, tzinfo=UTC + ) start_ts = int(start_dt.timestamp()) end_ts = int(end_dt.timestamp()) diff --git a/src/ml4t/data/providers/polygon.py b/src/ml4t/data/providers/polygon.py index db2e841c..b18b2875 100644 --- a/src/ml4t/data/providers/polygon.py +++ b/src/ml4t/data/providers/polygon.py @@ -338,7 +338,7 @@ def _fetch_raw_data( ) raise ProviderError(provider=self.name, message=f"API error: {error_msg}") - return data + return self._collect_aggregate_pages(data) except ( AuthenticationError, @@ -351,6 +351,27 @@ def _fetch_raw_data( except Exception as err: raise NetworkError(provider=self.name, message=f"Request failed: {endpoint}") from err + def _collect_aggregate_pages(self, data: dict[str, Any]) -> dict[str, Any]: + """Follow aggregate ``next_url`` cursors so a 50,000-row page is not the whole result.""" + next_url = data.get("next_url") + if not isinstance(next_url, str) or not next_url: + return data + raw_results = data.get("results") or [] + results = ( + [item for item in raw_results if isinstance(item, dict)] + if isinstance(raw_results, list) + else [] + ) + seen: set[str] = set() + while isinstance(next_url, str) and next_url and next_url not in seen: + seen.add(next_url) + page = self._get_json(next_url, {}) + results.extend(item for item in page.get("results", []) if isinstance(item, dict)) + next_url = page.get("next_url") + if seen: + return {**data, "results": results} + return data + def _transform_data(self, raw_data: dict[str, Any], symbol: str) -> pl.DataFrame: """Transform raw API response to Polars DataFrame.""" if not raw_data.get("results"): diff --git a/src/ml4t/data/providers/polymarket.py b/src/ml4t/data/providers/polymarket.py index d54e406e..d2673280 100644 --- a/src/ml4t/data/providers/polymarket.py +++ b/src/ml4t/data/providers/polymarket.py @@ -30,7 +30,7 @@ """ import json -from datetime import datetime, timedelta +from datetime import UTC, datetime, timedelta from math import isfinite from typing import Any, ClassVar @@ -378,8 +378,8 @@ def _fetch_price_history( List of price points [{t: timestamp, p: price}, ...] """ try: - start_dt = datetime.strptime(start, "%Y-%m-%d") - end_dt = datetime.strptime(end, "%Y-%m-%d") + start_dt = datetime.strptime(start, "%Y-%m-%d").replace(tzinfo=UTC) + end_dt = datetime.strptime(end, "%Y-%m-%d").replace(tzinfo=UTC) history: list[dict[str, Any]] = [] chunk_days = self.HISTORY_CHUNK_DAYS if interval != "max" else self.MAX_MARKET_SCAN diff --git a/src/ml4t/data/update_manager.py b/src/ml4t/data/update_manager.py index ce2d1fe3..f94b4723 100644 --- a/src/ml4t/data/update_manager.py +++ b/src/ml4t/data/update_manager.py @@ -195,8 +195,9 @@ def detect_gaps_in_storage( start_date = _ensure_datetime(start_date) end_date = _ensure_datetime(end_date) try: - # Read data from storage - df = storage.read(key, start_date, end_date).collect() + # Storage end bounds are exclusive. Open the inclusive end by one + # microsecond so a bar timestamped on end_date is part of the range. + df = storage.read(key, start_date, end_date + timedelta(microseconds=1)).collect() if df.is_empty(): # No data in range means entire range is a gap diff --git a/tests/test_fama_french_provider.py b/tests/test_fama_french_provider.py index 5fbffe98..6ef654cc 100644 --- a/tests/test_fama_french_provider.py +++ b/tests/test_fama_french_provider.py @@ -315,6 +315,29 @@ def test_fetch_date_filtering(self): assert len(df) == 2 # Only 2024 dates + def test_fetch_month_filter_includes_final_calendar_day(self): + """YYYY-MM end filters include observations after the 28th.""" + with tempfile.TemporaryDirectory() as tmpdir: + provider = FamaFrenchProvider(cache_path=tmpdir, use_cache=True) + cached_df = pl.DataFrame( + { + "timestamp": [ + datetime(2024, 2, 28), + datetime(2024, 2, 29), + datetime(2024, 3, 1), + ], + "Mkt-RF": [0.01, 0.02, 0.03], + } + ) + cached_df.write_parquet(Path(tmpdir) / "ff3_daily.parquet") + + filtered = provider.fetch("ff3", frequency="daily", start="2024-02", end="2024-02") + + assert filtered["timestamp"].to_list() == [ + datetime(2024, 2, 28), + datetime(2024, 2, 29), + ] + def test_fetch_combined(self, provider): """Test fetch_combined merges datasets.""" with tempfile.TemporaryDirectory() as tmpdir: diff --git a/tests/test_finnhub_provider.py b/tests/test_finnhub_provider.py index ed1e3f4d..a561281d 100644 --- a/tests/test_finnhub_provider.py +++ b/tests/test_finnhub_provider.py @@ -1,5 +1,6 @@ """Tests for Finnhub provider module.""" +from datetime import UTC, datetime from unittest.mock import MagicMock, patch import polars as pl @@ -252,6 +253,20 @@ def test_fetch_raw_data_error_status(self, provider): with pytest.raises(ProviderError, match="API error"): provider._fetch_raw_data("AAPL", "2024-01-01", "2024-01-02", "daily") + def test_fetch_raw_data_uses_inclusive_utc_day(self, provider): + """Date-only bounds are UTC days and include the end date.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"s": "no_data"} + + with patch.object(provider.session, "get", return_value=mock_response) as mock_get: + with pytest.raises(SymbolNotFoundError): + provider._fetch_raw_data("AAPL", "2024-01-31", "2024-01-31", "1min") + + params = mock_get.call_args.kwargs["params"] + assert params["from"] == int(datetime(2024, 1, 31, tzinfo=UTC).timestamp()) + assert params["to"] == int(datetime(2024, 1, 31, 23, 59, 59, tzinfo=UTC).timestamp()) + def test_fetch_raw_data_missing_data_arrays(self, provider): """Test missing data arrays raise SymbolNotFoundError.""" mock_response = MagicMock() diff --git a/tests/test_kalshi_unit.py b/tests/test_kalshi_unit.py index c2a3a44f..3c55309c 100644 --- a/tests/test_kalshi_unit.py +++ b/tests/test_kalshi_unit.py @@ -14,7 +14,7 @@ - List markets/series methods """ -from datetime import datetime +from datetime import UTC, datetime from unittest.mock import MagicMock, patch from urllib.parse import urlparse @@ -409,6 +409,19 @@ def provider(self): yield provider provider.close() + def test_fetch_raw_data_uses_utc_calendar_bounds(self, provider): + """Inclusive date bounds are converted in UTC, not the local timezone.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"candlesticks": []} + + with patch.object(provider.session, "get", return_value=mock_response) as mock_get: + provider._fetch_raw_data("KXINFL-25JAN", "2024-01-31", "2024-01-31", "daily") + + params = mock_get.call_args.kwargs["params"] + assert params["start_ts"] == int(datetime(2024, 1, 31, tzinfo=UTC).timestamp()) + assert params["end_ts"] == int(datetime(2024, 1, 31, 23, 59, 59, tzinfo=UTC).timestamp()) + def test_fetch_ohlcv_success(self, provider): """Test successful fetch_ohlcv with mocked response.""" ts1 = int(datetime(2024, 1, 2, 12, 0).timestamp()) diff --git a/tests/test_macro_downloader.py b/tests/test_macro_downloader.py index 884e5466..14f8616a 100644 --- a/tests/test_macro_downloader.py +++ b/tests/test_macro_downloader.py @@ -222,11 +222,12 @@ def test_download_from_yfinance(self, manager): ) mock_df.columns = pd.MultiIndex.from_tuples([("Close", "")]) - with patch("yfinance.download", return_value=mock_df): + with patch("yfinance.download", return_value=mock_df) as mock_download: df = manager._download_from_yfinance(["DGS10"]) assert not df.is_empty() assert "DGS10" in df.columns + assert mock_download.call_args.kwargs["end"] == "2025-01-01" def test_compute_derived_series(self, manager): """Test _compute_derived_series method.""" diff --git a/tests/test_polygon_provider.py b/tests/test_polygon_provider.py index 4c2f75e3..a6707cf3 100644 --- a/tests/test_polygon_provider.py +++ b/tests/test_polygon_provider.py @@ -386,6 +386,30 @@ def test_api_parameters(self): assert params.get("limit") == 50000 assert params.get("apiKey") == "test_key" + def test_fetch_raw_data_follows_next_url(self): + """Aggregate responses continue through next_url instead of stopping at one page.""" + provider = MassiveProvider(api_key="test_key") + first = MagicMock() + first.status_code = 200 + first.json.return_value = { + "status": "OK", + "results": [{"t": 1, "o": 1.0, "h": 1.0, "l": 1.0, "c": 1.0, "v": 1.0}], + "next_url": "https://api.massive.com/v2/aggs/next?cursor=abc", + } + second = MagicMock() + second.status_code = 200 + second.json.return_value = { + "status": "OK", + "results": [{"t": 2, "o": 2.0, "h": 2.0, "l": 2.0, "c": 2.0, "v": 2.0}], + } + + with patch.object(provider.session, "get", side_effect=[first, second]) as mock_get: + data = provider._fetch_raw_data("AAPL", "2024-01-01", "2024-03-01", "daily") + + assert [row["t"] for row in data["results"]] == [1, 2] + assert mock_get.call_count == 2 + assert mock_get.call_args_list[1].kwargs["params"]["apiKey"] == "test_key" + class TestMassiveAssetClassRouting: """Tests for Massive endpoint routing across asset classes.""" diff --git a/tests/test_polymarket_unit.py b/tests/test_polymarket_unit.py index 3e5451da..e27aa49b 100644 --- a/tests/test_polymarket_unit.py +++ b/tests/test_polymarket_unit.py @@ -3,7 +3,7 @@ These tests verify the Polymarket provider logic without making real API calls. """ -from datetime import datetime +from datetime import UTC, datetime from unittest.mock import MagicMock, patch import polars as pl @@ -392,6 +392,17 @@ def test_fetch_price_history_chunks_long_ranges(self, mock_fetch_chunk, provider assert mock_fetch_chunk.call_count == 2 assert [point["t"] for point in history] == [1704067200, 1704153600, 1704240000] + @patch.object(PolymarketProvider, "_fetch_price_history_chunk") + def test_fetch_price_history_uses_utc_day_bounds(self, mock_fetch_chunk, provider): + """Price-history windows are inclusive UTC days.""" + mock_fetch_chunk.return_value = [] + + provider._fetch_price_history("12345678901234567890", "2024-01-31", "2024-01-31", "1d") + + _token_id, start_ts, end_ts, _interval = mock_fetch_chunk.call_args.args + assert start_ts == int(datetime(2024, 1, 31, tzinfo=UTC).timestamp()) + assert end_ts == int(datetime(2024, 1, 31, 23, 59, 59, tzinfo=UTC).timestamp()) + @patch.object(PolymarketProvider, "_fetch_price_history_chunk") def test_fetch_price_history_normalizes_numeric_timestamps(self, mock_fetch_chunk, provider): """Retain numeric JSON timestamps and report malformed entries.""" diff --git a/tests/test_update_manager.py b/tests/test_update_manager.py index 2f930e7e..a02e8bde 100644 --- a/tests/test_update_manager.py +++ b/tests/test_update_manager.py @@ -320,6 +320,24 @@ def test_detect_gaps_in_storage_with_data( # Should detect gap at the end (Feb 1-15) assert len(gaps) >= 1 + def test_detect_gaps_in_storage_keeps_bar_on_inclusive_end(self, storage: HiveStorage) -> None: + """A bar timestamped on the requested end date is inside the range.""" + detector = GapDetector() + dates = [datetime(2024, 1, 1) + timedelta(days=i) for i in range(10)] + storage.write( + pl.DataFrame({"timestamp": dates, "close": [1.0] * len(dates)}), + "test_symbol", + ) + + gaps = detector.detect_gaps_in_storage( + storage, + "test_symbol", + datetime(2024, 1, 1), + datetime(2024, 1, 10), + ) + + assert gaps == [] + class TestIncrementalUpdater: """Test IncrementalUpdater for update strategies."""