fix(indicators): volume statistics carry the unit their source declared (#1571)

#1571 added latest / 20-bar mean / ratio volume and chose not to infer a unit.
The pipeline does not ask for inference, though. fetch_market_data's contract
(#1062) is that consumers read provenance.volume_unit instead of assuming one:
the A-share sources count board lots of 100 and the Yahoo family counts shares,
so an unlabelled latest volume can be read 100x wrong. The tool now requests
provenance and returns volume.unit as the serving loader declared it (lots /
shares), or null when undeclared. ratio_20 was already unit-free.

Also folds the new _extract_volume_series into the close extractor it copied,
about 30 lines, as one _extract_series(keys, keep_missing=). Every existing
test passes unchanged. keep_missing had no test that could fail: with dates,
the tool reindexes volume onto the close dates, so a gap comes back as NaN
either way. It matters only for an undated payload, where dropping the gap
shifted bar 28's volume into 'latest'. That case is now tested. Mutants (no
provenance / gaps dropped) each fail.
This commit is contained in:
Haozhe Wu committed 2026-09-24 18:43:23 +08:00
1 parent 71084c4173
commit 234a2e3ea8
2 files changed
+70 -59

No files matched your search

+38 -59
View File
@@ -42,13 +42,24 @@ _VOLUME_KEYS = ("volume", "Volume", "VOLUME")
_DATE_KEYS = ("trade_date", "date", "datetime", "timestamp", "time")
def _extract_close_series(payload: Any) -> pd.Series | None:
"""Extract the close-price series from a loader payload.
def _extract_series(
payload: Any, keys: tuple[str, ...], *, keep_missing: bool = False
) -> pd.Series | None:
"""Extract one numeric column from a loader payload, on its dates.
Loader results can be a ``DataFrame``, a plain column mapping, or the
``list[dict]`` shape returned by :func:`fetch_market_data`. Preserve a real
date index when one is available; a payload without dates intentionally
keeps a ``RangeIndex`` so callers do not report a row number as a date.
Args:
payload: The loader result for one symbol.
keys: Column spellings to look for, first match wins.
keep_missing: Keep a dated bar whose value is missing as NaN instead
of dropping it, so a gap cannot slide an older value into "latest".
Returns:
A float series, or ``None`` when the payload has no such column.
"""
if isinstance(payload, list):
if not payload or not isinstance(payload[0], dict):
@@ -57,7 +68,7 @@ def _extract_close_series(payload: Any) -> pd.Series | None:
elif isinstance(payload, dict):
inner = payload.get("data")
if isinstance(inner, list):
return _extract_close_series(inner)
return _extract_series(inner, keys, keep_missing=keep_missing)
try:
frame = pd.DataFrame(payload)
except ValueError:
@@ -67,11 +78,11 @@ def _extract_close_series(payload: Any) -> pd.Series | None:
else:
return None
close_key = next((key for key in _CLOSE_KEYS if key in frame.columns), None)
if close_key is None:
key = next((key for key in keys if key in frame.columns), None)
if key is None:
return None
close = pd.to_numeric(frame[close_key], errors="coerce")
values = pd.to_numeric(frame[key], errors="coerce")
date_key = next((key for key in _DATE_KEYS if key in frame.columns), None)
if date_key is not None:
dates = pd.to_datetime(frame[date_key], errors="coerce")
@@ -84,67 +95,26 @@ def _extract_close_series(payload: Any) -> pd.Series | None:
pd.to_datetime(frame.index, errors="coerce"), index=frame.index
)
else:
return close.dropna().reset_index(drop=True).astype(float)
kept = values if keep_missing else values.dropna()
return kept.reset_index(drop=True).astype(float)
valid = close.notna() & dates.notna()
valid = dates.notna() if keep_missing else values.notna() & dates.notna()
normalized = pd.Series(
close.loc[valid].to_numpy(dtype=float),
values.loc[valid].to_numpy(dtype=float),
index=pd.DatetimeIndex(dates.loc[valid]),
dtype=float,
)
return normalized.sort_index(kind="stable")
def _extract_close_series(payload: Any) -> pd.Series | None:
"""Extract the close-price series from a loader payload."""
return _extract_series(payload, _CLOSE_KEYS)
def _extract_volume_series(payload: Any) -> pd.Series | None:
"""Extract the volume series from a loader payload.
Missing or non-numeric volume values are preserved as NaN when a date
index is available so callers do not silently carry a stale observation
forward to the latest bar.
"""
if isinstance(payload, list):
if not payload or not isinstance(payload[0], dict):
return None
frame = pd.DataFrame(payload)
elif isinstance(payload, dict):
inner = payload.get("data")
if isinstance(inner, list):
return _extract_volume_series(inner)
try:
frame = pd.DataFrame(payload)
except ValueError:
return None
elif isinstance(payload, pd.DataFrame):
frame = payload.copy()
else:
return None
volume_key = next((key for key in _VOLUME_KEYS if key in frame.columns), None)
if volume_key is None:
return None
volume = pd.to_numeric(frame[volume_key], errors="coerce")
date_key = next((key for key in _DATE_KEYS if key in frame.columns), None)
if date_key is not None:
dates = pd.to_datetime(frame[date_key], errors="coerce")
elif isinstance(frame.index, pd.DatetimeIndex):
dates = pd.Series(frame.index, index=frame.index)
elif not isinstance(
frame.index, pd.RangeIndex
) and not pd.api.types.is_integer_dtype(frame.index.dtype):
dates = pd.Series(
pd.to_datetime(frame.index, errors="coerce"), index=frame.index
)
else:
return volume.reset_index(drop=True).astype(float)
valid_dates = dates.notna()
normalized = pd.Series(
volume.loc[valid_dates].to_numpy(dtype=float),
index=pd.DatetimeIndex(dates.loc[valid_dates]),
dtype=float,
)
return normalized.sort_index(kind="stable")
"""Extract the volume series, keeping a dated bar with no volume as NaN."""
return _extract_series(payload, _VOLUME_KEYS, keep_missing=True)
def _compute_volume_stats(
@@ -257,7 +227,9 @@ class TechnicalIndicatorTool(BaseTool):
name = "technical_indicators"
description = (
"Compute common technical indicators (RSI, MACD, Bollinger Bands, "
"SMA, EMA) plus latest/20-bar volume statistics for a trading symbol. "
"SMA, EMA) plus latest/20-bar volume statistics for a trading symbol "
"(volume.unit is the serving source's declared unit -- lots or shares -- "
"or null when undeclared). "
"Uses the project's data loaders "
"to fetch price history, then computes indicators locally."
)
@@ -328,6 +300,9 @@ class TechnicalIndicatorTool(BaseTool):
# Indicators require consecutive bars. ``max_rows=lookback``
# would make the shared helper even-stride sample long windows.
max_rows=0,
# For volume_unit: lots on the A-share sources, shares on the
# Yahoo family (#1062) -- 100x apart, and nothing to infer from.
include_provenance=True,
)
except Exception as exc:
logger.debug("fetch_market_data failed for %s: %s", symbol, exc)
@@ -369,6 +344,10 @@ class TechnicalIndicatorTool(BaseTool):
if volume is not None
else {"latest": None, "sma_20": None, "ratio_20": None}
)
provenance = (data.get("_provenance") or {}).get(symbol)
volume_stats["unit"] = (
provenance.get("volume_unit") if isinstance(provenance, dict) else None
)
# ── Compute indicators ────────────────────────────────────────────
indicators: dict[str, Any] = {
@@ -263,6 +263,7 @@ class TestTechnicalIndicatorToolIntegration:
"latest": None,
"sma_20": None,
"ratio_20": None,
"unit": None,
}
def test_execute_dataframe_with_adj_close(self, monkeypatch, sample_close):
@@ -367,6 +368,37 @@ class TestLoaderPayloadShapes:
1_000_029.0 / 1_000_019.5
)
def test_an_undated_bar_with_no_volume_is_not_filled_by_the_bar_before(self, monkeypatch):
"""Without dates, bars align by position: dropping the gap would report bar 28 as latest."""
records = self._records()
for record in records:
del record["trade_date"]
records[-1]["volume"] = None
monkeypatch.setattr(
"src.tools.technical_indicator_tool.fetch_market_data",
lambda **kw: {"MSFT": pd.DataFrame(records)},
)
result = json.loads(TechnicalIndicatorTool().execute(symbol="MSFT"))
assert result["indicators"]["volume"]["latest"] is None
assert result["indicators"]["volume"]["sma_20"] is None
@pytest.mark.parametrize("declared", ["lots", "shares", None])
def test_volume_carries_the_unit_its_source_declared(self, monkeypatch, declared):
"""An A-share source counts board lots of 100, the Yahoo family shares (#1062)."""
seen: dict = {}
provenance = {} if declared is None else {"600519.SH": {"volume_unit": declared}}
def fetch(**kwargs):
seen.update(kwargs)
return {"600519.SH": self._records(), "_provenance": provenance}
monkeypatch.setattr("src.tools.technical_indicator_tool.fetch_market_data", fetch)
result = json.loads(TechnicalIndicatorTool().execute(symbol="600519.SH"))
assert seen["include_provenance"] is True
assert result["indicators"]["volume"]["unit"] == declared
def test_wrapped_capped_payload_is_rejected(self, monkeypatch):
"""A discontinuous capped payload must never produce indicators."""
payload = {