Green ruff lint (supersedes #31): databento start_date + ruff-clean

.github/workflows/python-test.yml

@@ -24,6 +24,8 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         python -m pip install -e ".[dev]"
+    - name: Run linting
+      run: python -m ruff check src tests
     - name: Run tests
       run: |
         # Create a dummy store for the test environment to prevent COTDATA_STORE missing errors

CHANGELOG.md

@@ -25,6 +25,21 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   generalization markets. ([#24](https://github.com/mspinola/cotdata/pull/24))
 
 ### Fixed
+- **`databento.fetch_daily_ohlc`'s `start_date` parameter now actually does
+  something** (dormant provider). It was previously silently ignored on every
+  code path — a cold cache always backfilled from 2000-01-01 and a warm cache
+  always resumed from `last_date + 1 day` regardless of what was passed — so a
+  caller trying to bound a fetch (e.g. "just the last 3 months") got a
+  full-history pull instead, at full Databento API cost, with no error or
+  warning. Now: on a **cold** cache (a symbol's first-ever fetch) `start_date`
+  narrows the fetch floor, so a narrow first-time query is actually cheap. On
+  a **warm** cache it does *not* narrow the incremental fetch — the top-up
+  always resumes from the cache's own `last_date + 1`, so a later, narrower
+  `start_date` can never silently truncate a cache other callers already rely
+  on being complete — but the *returned* frame is still filtered to
+  `>= start_date` either way. The on-disk cache always persists the full
+  series regardless of `start_date`; only the fetch cost (cold cache) and the
+  return value (always) are affected.
 - **Fail fast when the Norgate service (NDU) is unreachable** — the producer now
   probes `norgatedata.status()` before fetching and aborts with a clear error and
   a non-zero exit. Previously norgatedata retried each call 10x then called bare

pyproject.toml

@@ -38,7 +38,7 @@ norgate = ["norgatedata"]
 databento = ["databento>=0.30"]
 
 yahoo = ["yfinance"]
-dev = ["pytest>=7"]
+dev = ["pytest>=7", "ruff==0.15.22"]
 
 [project.scripts]
 cotdata-update = "cotdata.update:main"
@@ -48,3 +48,11 @@ where = ["src"]
 
 [tool.setuptools.package-data]
 cotdata = ["*.yaml"]
+
+[tool.ruff]
+line-length = 100
+target-version = "py39"
+
+[tool.ruff.lint]
+select = ["E", "F", "W", "I"]
+ignore = ["E501"]

src/cotdata/__init__.py

@@ -1,8 +1,8 @@
 """cotdata — canonical data layer (see README). Public consumer API."""
-from .prices import get_prices, roll_dates
 from .cot import get_cot
-from .registry import symbol, all_symbols, REGISTRY, Symbol
-from .store import load_manifest, schema_version, require_schema
+from .prices import get_prices, roll_dates
+from .registry import REGISTRY, Symbol, all_symbols, symbol
+from .store import load_manifest, require_schema, schema_version
 
 __version__ = "0.1.0"
 __all__ = [

src/cotdata/cot.py

@@ -28,7 +28,7 @@ def get_cot(name: str, report: str = "legacy") -> pd.DataFrame:
     sym = REGISTRY.get(name)
     if sym is None:                       # allow lookup by primary CFTC code, not just symbol
         sym = next((s for s in REGISTRY.values() if s.cftc_code == name), None)
-    
+
     read_fn = {
         "legacy": store.read_cot_legacy,
         "disagg": store.read_cot_disagg,

src/cotdata/providers/cftc_disagg.py

@@ -5,7 +5,6 @@
 the store via store.write_cot_disagg().
 """
 import datetime as dt
-import io
 import zipfile
 from email.utils import parsedate_to_datetime
 from pathlib import Path
@@ -60,27 +59,27 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame:
         with zf.open(zf.namelist()[0]) as fh:
             # The .txt files are actually CSVs. low_memory=False prevents dtype warnings.
             df = pd.read_csv(fh, low_memory=False)
-            
+
     # Strip any trailing whitespace from column names BEFORE accessing them
     df.columns = df.columns.str.strip()
-    
+
     # CFTC sometimes changes the date column name in Disaggregated reports
     if REPORT_DATE not in df.columns:
         date_cols = [c for c in df.columns if "Report_Date" in c]
         if date_cols:
             df.rename(columns={date_cols[0]: REPORT_DATE}, inplace=True)
         else:
             print(f"AVAILABLE COLUMNS: {list(df.columns)}")
-            
+
     # Coerce the key schema columns to match the Legacy schema format
     df[CONTRACT_CODE] = df[CONTRACT_CODE].apply(_standardize_code)
     df[REPORT_DATE] = pd.to_datetime(df[REPORT_DATE]).dt.tz_localize(None)
-    
+
     # Parquet cannot serialize mixed-type object columns (e.g. Traders_Tot_Old contains ints and strings)
     for col in df.select_dtypes(include=['object']).columns:
         if col not in [CONTRACT_CODE, REPORT_DATE]:
             df[col] = df[col].astype(str)
-            
+
     return df
 
 
@@ -105,7 +104,7 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> dict:
         want = set(code_to_sym.keys())
 
     frames = []
-    
+
     # The CFTC bundles 2006-2016 in a single historical file
     if first_year <= 2016:
         url = "https://www.cftc.gov/files/dea/history/fut_disagg_txt_hist_2006_2016.zip"

src/cotdata/providers/cftc_tff.py

@@ -5,7 +5,6 @@
 the store via store.write_cot_tff().
 """
 import datetime as dt
-import io
 import zipfile
 from email.utils import parsedate_to_datetime
 from pathlib import Path
@@ -60,27 +59,27 @@ def _parse_zip(zip_path: Path) -> pd.DataFrame:
         with zf.open(zf.namelist()[0]) as fh:
             # The .txt files are actually CSVs. low_memory=False prevents dtype warnings.
             df = pd.read_csv(fh, low_memory=False)
-            
+
     # Strip any trailing whitespace from column names BEFORE accessing them
     df.columns = df.columns.str.strip()
-    
+
     # CFTC sometimes changes the date column name in TFF reports
     if REPORT_DATE not in df.columns:
         date_cols = [c for c in df.columns if "Report_Date" in c]
         if date_cols:
             df.rename(columns={date_cols[0]: REPORT_DATE}, inplace=True)
         else:
             print(f"AVAILABLE COLUMNS: {list(df.columns)}")
-            
+
     # Coerce the key schema columns to match the Legacy schema format
     df[CONTRACT_CODE] = df[CONTRACT_CODE].apply(_standardize_code)
     df[REPORT_DATE] = pd.to_datetime(df[REPORT_DATE]).dt.tz_localize(None)
-    
+
     # Parquet cannot serialize mixed-type object columns (e.g. Traders_Tot_Old contains ints and strings)
     for col in df.select_dtypes(include=['object']).columns:
         if col not in [CONTRACT_CODE, REPORT_DATE]:
             df[col] = df[col].astype(str)
-            
+
     return df
 
 
@@ -105,7 +104,7 @@ def update(codes=None, first_year: int = FIRST_YEAR, last_year=None) -> dict:
         want = set(code_to_sym.keys())
 
     frames = []
-    
+
     # The CFTC bundles 2006-2016 in a single historical file
     if first_year <= 2016:
         url = "https://www.cftc.gov/files/dea/history/fin_fut_txt_2006_2016.zip"

src/cotdata/providers/databento.py

@@ -22,6 +22,7 @@
 import time
 import warnings
 from pathlib import Path
+from typing import Optional
 
 import pandas as pd
 
@@ -114,11 +115,41 @@ def run_batch_backfill(symbols: list) -> None:
                 df_clean.to_parquet(_cache_dir() / f"{internal_sym}_daily.parquet")
 
 
-def fetch_daily_ohlc(symbol: str, start_date: str = "2000-01-01",
+def fetch_daily_ohlc(symbol: str, start_date: Optional[str] = None,
                      force_refresh: bool = False, price_type: str = "close") -> pd.DataFrame:
     """Daily OHLC + Open Interest via Databento (GLBX.MDP3 ohlcv-1d + statistics),
     append-only cache; yfinance fallback for _DATABENTO_UNSUPPORTED. price_type
-    'settlement' pulls stat_type 3 dated by ts_ref."""
+    'settlement' pulls stat_type 3 dated by ts_ref.
+
+    The cache is always a single, from-inception, append-only series per
+    symbol+price_type, shared across every caller — that is what makes repeated
+    calls cheap. `start_date` therefore means two different things depending on
+    whether it is the FIRST call for a symbol or not, and both are deliberate:
+
+    * **Cold cache (symbol never fetched before):** `start_date` clamps the fetch
+      floor (still no lower than the GLBX.MDP3 history floor 2010-06-06), so a
+      narrow first-time query — e.g. "just the last 3 months" — actually costs a
+      narrow API pull, not a from-2000 one.
+    * **Warm cache (symbol already has some history):** `start_date` does NOT
+      change what gets fetched — the top-up always resumes from the cache's
+      `last_date + 1 day`, exactly as before. Letting a later, narrower
+      `start_date` shrink the fetch would silently truncate a cache other
+      callers already rely on being complete. `start_date` still filters what is
+      RETURNED to this call, just not what is fetched or persisted.
+
+    In both cases the returned frame never contains rows before `start_date` — the
+    filter is applied uniformly to whatever the cache ends up holding. `None`
+    (the default) is unbounded: full cached history, unfiltered, unchanged
+    behaviour from before this parameter existed.
+
+    One consequence of the "warm cache never shrinks its floor" rule: if a symbol's
+    cache was FIRST populated by a narrow `start_date` (e.g. "since 2024"), a later
+    call asking for `start_date="2010-01-01"` will NOT backfill 2010-2023 — the
+    cache only ever grows forward. Not a concern for the one real caller today
+    (`update_all_daily_prices`, which never passes `start_date` and always wants
+    full history), but worth knowing before relying on this for a symbol multiple
+    call sites touch with different windows.
+    """
     import databento as db
     display_sym = symbol
     _pt_suffix = "" if price_type == "close" else f"_{price_type}"
@@ -135,24 +166,33 @@ def fetch_daily_ohlc(symbol: str, start_date: str = "2000-01-01",
         except Exception as e:  # noqa: BLE001
             logger.warning("Failed to read cache for %s: %s", display_sym, e)
 
+    def _filtered(df: pd.DataFrame) -> pd.DataFrame:
+        if start_date and not df.empty:
+            return df[df.index >= pd.Timestamp(start_date)]
+        return df
+
     if local_df.empty:
-        fetch_start = "2000-01-01"
+        # Cold cache: start_date narrows the fetch floor (still API-cost-bounded
+        # by the GLBX floor below), so a first-time narrow query is actually cheap.
+        fetch_start = start_date if start_date else "2000-01-01"
     else:
+        # Warm cache: start_date does NOT narrow this — see the docstring. It only
+        # affects the return-value filter below.
         last_date = local_df.index.max()
         today = pd.Timestamp.now().normalize()
         if last_date >= today - pd.Timedelta(days=1):
-            return local_df
+            return _filtered(local_df)
         if "--fast" in sys.argv:
-            return local_df
+            return _filtered(local_df)
         now = time.time()
         if not force_refresh and (now - _API_LAST_CHECKED.get(symbol, 0) < 3600):
-            return local_df
+            return _filtered(local_df)
         _API_LAST_CHECKED[symbol] = now
         fetch_start = (last_date + pd.Timedelta(days=1)).strftime("%Y-%m-%d")
 
     fetch_end = (pd.Timestamp.now() - pd.Timedelta(days=1)).strftime("%Y-%m-%d")
     if pd.Timestamp(fetch_start) >= pd.Timestamp(fetch_end):
-        return local_df
+        return _filtered(local_df)
 
     new_df = pd.DataFrame()
     db_key = os.environ.get("DATABENTO_API_KEY")
@@ -245,12 +285,12 @@ def fetch_daily_ohlc(symbol: str, start_date: str = "2000-01-01",
     if not local_df.empty and not new_df.empty:
         new_df = new_df[new_df.index > local_df.index.max()]
     if new_df.empty:
-        return local_df
+        return _filtered(local_df)
     new_df = new_df.dropna(subset=["Close"])
     combined = pd.concat([local_df, new_df]) if not local_df.empty else new_df
     combined = combined[~combined.index.duplicated(keep="last")].sort_index()
-    combined.to_parquet(cache_path)
-    return combined
+    combined.to_parquet(cache_path)          # cache always persists the FULL series
+    return _filtered(combined)               # start_date only shapes what's returned
 
 
 def update_all_daily_prices(force_refresh: bool = False) -> None: