From 7b7c427076b86c194d27f9bbee305760a0e85234 Mon Sep 17 00:00:00 2001 From: Matt Spinola Date: Sun, 2 Aug 2026 16:50:47 -0400 Subject: [PATCH] refactor(vintage): drop the flow decomposition, which was a less capable copy vintage_flow.decompose and crowdmon.futures.flow.decompose were carried side by side since 2026-08-01 as "defensible alternatives answering slightly different questions". Measured, they are one function: cotdata.vintage_flow.decompose == crowdmon.futures.flow.decompose at tolerance=1.0, gap rule off On 27 markets and 135,835 transitions from 2006 to 2026: identical row sets, 100.000000% label agreement, zero mismatches, and d_long/d_short/d_net identical on every row. Both take the dominant leg as argmax(|dLong|,|dShort|), both break exact ties to the long leg, and both treat a doubly-unmoved week as quiet unconditionally, so setting tolerance=1.0 (nothing is ever mixed) reproduces this classifier exactly. This copy was therefore not a second opinion but the less capable one. It could not decline to label a genuinely two-sided week, and it differenced across a 294-day absence as though it were a week. The general implementation stays in crowdmon, where the module spec puts the positioning engine. The dedup could not go the other way: crowdmon/tests/test_boundaries.py forbids cotdata from importing crowdmon, since a producer depending on its consumer inverts the dependency direction. The equivalence is asserted from that side, in crowdmon/tests/test_flow_equivalence.py, and written up in crowdmon/docs/design/amendments-2026-08-02.md B29. NO DEPRECATION PATH, and the reason is checkable rather than asserted. decompose was never in cotdata.__all__, has never appeared in CHANGELOG.md under any version, and is in no published release: PyPI carries 0.1.0, which predates the whole vintage subsystem, and the 0.2.0 note says in terms that tagging is not publishing. Its only caller was the CLI subcommand removed with it. KEPT. zero_sum_check, from_vintage and from_current_store stay. The zero-sum identity is a claim about cotdata's own parse, and crowdmon.futures.cot_adapter runs it on every load. The `flow` subcommand becomes `zero-sum`, keeping the half that was cotdata's own and adding the rounding-tolerance count and a report of non-weekly intervals (COT was fortnightly before 1992-10-13); a command named `flow` that no longer decomposes flow would be worse than a renamed one. DROPPED DELIBERATELY. min_frac_oi, the optional dead zone, has no equivalent in the surviving implementation. It defaulted to 0.0, nothing set it, and the dominance tolerance answers the same problem with a sensitivity sweep attached. docs/design/cot_vintage.md section 7 records where that argument turned out to be half right: parameter-free was a real virtue and it cost the ability to say a week was two-sided, which is the modal outcome at 60% of weeks. Section 7 is kept rather than deleted. What it records is a result about the canonical SCHEMA, obtained by running a decomposition over it, and the zero-sum sweep, the spreading finding and the validate() finding do not depend on where the classifier lives. 262 tests pass, ruff clean. crowdmon's own suite passes against this branch (433 passed, its four equivalence tests skipping with a stated reason) and against current main (437 passed, equivalence enforced), so the two can merge in either order without a red build. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 30 +++++ docs/design/cot_vintage.md | 34 +++++- src/cotdata/vintage_cli.py | 71 ++++++------ src/cotdata/vintage_flow.py | 208 +++++++--------------------------- src/cotdata/vintage_ingest.py | 3 +- tests/test_vintage_flow.py | 119 +++++-------------- 6 files changed, 178 insertions(+), 287 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c0d2a7b..2f99f99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,36 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Removed +- **`vintage_flow.decompose` and the `cotdata-vintage flow` subcommand.** Flow + decomposition was duplicated in `crowdmon.futures.flow`, and measurement showed the two + were not alternatives but **one function**: this copy was that one at `tolerance=1.0` + with the gap rule off, agreeing on **100.000000% of 135,835 transitions (2006-2026, 27 + markets) with zero mismatches**, and `d_long`/`d_short`/`d_net` identical on every row. + The copy here was strictly the less capable one: it could not decline to label a + genuinely two-sided week (no `mixed` state) and it differenced across a 294-day absence + as though it were a week. The general implementation stays in `crowdmon`, which is where + the module spec puts the positioning engine. The dedup could not go the other way, + because `crowdmon/tests/test_boundaries.py` forbids cotdata from importing its own + consumer. Measurement in `crowdmon/docs/design/amendments-2026-08-02.md` §B29, asserted + by `crowdmon/tests/test_flow_equivalence.py`. + + **No deprecation path, and the reason is checkable rather than asserted.** + `decompose` was never in `cotdata.__all__`, has never appeared in this changelog under + any version, and is in no **published** release: PyPI carries 0.1.0, which predates the + entire vintage subsystem, and the 0.2.0 note below states in terms that tagging is not + publishing. Its only caller in the workspace was the CLI subcommand removed with it. + + **Not removed:** `zero_sum_check`, `from_vintage` and `from_current_store` stay. The + zero-sum identity is a statement about cotdata's own parse and is consumed by + `crowdmon.futures.cot_adapter` on every load. `min_frac_oi`, the optional dead zone, has + no equivalent in the surviving implementation and is gone; it defaulted to off and + nothing set it. +- **`cotdata-vintage flow` is replaced by `cotdata-vintage zero-sum`**, which keeps the + half of that command that was cotdata's own: source selection plus the identity check, + now also reporting the rounding-tolerance column and any non-weekly intervals. A command + named `flow` that no longer decomposes flow would be a worse outcome than a renamed one. + ## [0.2.0] - 2026-08-02 The vintage release. `cotdata` gains an as-published provenance layer beside the diff --git a/docs/design/cot_vintage.md b/docs/design/cot_vintage.md index 2f375df..8e72b1c 100644 --- a/docs/design/cot_vintage.md +++ b/docs/design/cot_vintage.md @@ -359,7 +359,27 @@ Two smaller things fell out: ## 7. Flow decomposition, and what it found in the canonical schema -Module spec §6.4 built as `vintage_flow.py`. Weekly ΔLong versus ΔShort per +> **The decomposition has been REMOVED from this package (2026-08-02). What it found has +> not.** `crowdmon.futures.flow.decompose` was built independently on 2026-08-01, and the +> two were carried side by side as defensible alternatives until somebody measured them: +> this one was that one **at `tolerance=1.0` with the gap rule off**, agreeing on +> **100.000000% of 135,835 transitions with zero mismatches** and identical `d_long`, +> `d_short` and `d_net` on every row. One function, with the copy here hard-wired to the +> corner where nothing is ever `mixed` and no interval is ever refused. The measurement is +> in `crowdmon/docs/design/amendments-2026-08-02.md` §B29 and is asserted by +> `crowdmon/tests/test_flow_equivalence.py`. +> +> **This section is kept rather than deleted** because what it records is a result about +> *the canonical schema*, obtained by running a decomposition over it. The zero-sum sweep, +> the spreading finding and the `validate()` finding below all stand and none of them +> depend on where the classifier lives. `zero_sum_check` stays here; it is a statement +> about cotdata's own parse. +> +> The design points immediately below are preserved as the record of how the classifier +> was arrived at. **They describe `crowdmon.futures.flow` now**, with one exception noted +> inline. + +Module spec §6.4 was built as `vintage_flow.py`. Weekly ΔLong versus ΔShort per market/category, labelled `new_longs` / `short_covering` / `new_shorts` / `long_liquidation`. No prices, no contract master, no multiplier: every input is a column the canonical schema already stores, which is what makes it a clean smoke test of that @@ -374,6 +394,18 @@ Two design points that are not in the spec because they only appear against real to overfit. An optional `min_frac_oi` dead zone adds a `quiet` state; it defaults to 0.0 because any other value is a judgement in the same class as the fragility weights and belongs in a caller's config where it can be swept. + + > **This is the one point that did not carry over, and the argument turned out to be + > half right.** Parameter-free was a real virtue and it cost the ability to say a week + > was genuinely two-sided: dominant-leg always commits, so a category that added 30,000 + > longs and 28,000 shorts is called `new_longs`. The surviving implementation takes a + > dominance `tolerance` instead and reports a `mixed` state, which is the modal outcome + > at **60% of weeks** on the liquid panel, so the four-state table describes a minority + > of the data. The objection that a parameter is something to overfit was answered by + > sweeping it rather than by avoiding it: the tolerance moves 28.74% of labels across + > 0.15 to 0.40, and **not one of those changes moves a week from one pure state to a + > different pure state**. It gates whether the classifier commits, never which direction + > it commits to. `min_frac_oi` itself was never set by anything and is simply gone. - **`oi_corroborates`.** Futures are closed and zero-sum, so contracts exist only because somebody opened them: fresh positioning should coincide with rising open interest and exits with falling. Where it does not, the label is describing a transfer of an existing diff --git a/src/cotdata/vintage_cli.py b/src/cotdata/vintage_cli.py index 436f871..44768ed 100644 --- a/src/cotdata/vintage_cli.py +++ b/src/cotdata/vintage_cli.py @@ -270,41 +270,51 @@ def _cmd_asof(args) -> int: return 0 -def _cmd_flow(args) -> int: +def _cmd_zero_sum(args) -> int: + """The schema check that was the first half of `flow`. + + `flow` also printed a decomposition, which has moved to `crowdmon.futures.flow` (it was + the same function there, hard-wired to one corner of its parameters). What is left is + cotdata's own claim about its own parse, so the subcommand is named for that instead of + keeping a name that would now be a lie. + """ from . import vintage_flow if args.source == "current": canonical = vintage_flow.from_current_store(args.market) - print(f"vintage flow: {args.market} from the CURRENT-STATE store. NOT " - f"point-in-time. Revisions are already applied, so this is for looking at " - f"history and validating the schema, never for evaluating a rule.") + print(f"zero-sum: {args.market} from the CURRENT-STATE store. NOT point-in-time. " + f"Revisions are already applied, so this is for looking at history and " + f"validating the schema, never for evaluating a rule.") else: canonical = vintage_flow.from_vintage(as_of=args.as_of, market_code=args.market) if canonical.empty: - print("vintage flow: no vintage observations yet. The vintage series begins " - "at first capture; use --source current for history.") + print("zero-sum: no vintage observations yet. The vintage series begins at " + "first capture; use --source current for history.") return 0 z = vintage_flow.zero_sum_check(canonical) unbalanced = int((~z["balanced"]).sum()) - print(f" zero-sum: {len(z) - unbalanced}/{len(z)} weeks balanced" + outside = int((~z["within_tolerance"]).sum()) + print(f" {len(z) - unbalanced}/{len(z)} weeks balanced exactly" + (f" *** {unbalanced} UNBALANCED ***" if unbalanced else "")) - - fl = vintage_flow.decompose(canonical, min_frac_oi=args.min_frac_oi) - if args.category: - fl = fl[fl["category"] == args.category] - if fl.empty: - print(" no weeks to decompose.") - return 0 - off = fl[fl["days_elapsed"] != 7] + print(f" {len(z) - outside}/{len(z)} within CFTC's own rounding tolerance" + + (f" *** {outside} OUTSIDE ***" if outside else "")) + + # Not a weekly interval is a real finding about the series, so it is reported here + # rather than lost with the decomposition: COT was FORTNIGHTLY before 1992-10-13, and + # holidays shift the rest. A consumer differencing these rows must handle it. + dates = z["report_date"].sort_values().drop_duplicates() + gaps = dates.diff().dt.days.dropna() + off = gaps[gaps != 7] if len(off): - print(f" {len(off)} of {len(fl)} intervals are not 7 days (COT was FORTNIGHTLY " - f"before 1992-10-13, and holidays shift the rest). Those rows are not a " - f"weekly change and are not comparable to one.") - print(f"\n{fl['state'].value_counts().to_string()}\n") - cols = ["report_date", "category", "d_long", "d_short", "d_net", "d_oi", - "days_elapsed", "state", "oi_corroborates"] - print(fl.tail(args.last)[cols].to_string(index=False)) + print(f" {len(off)} of {len(gaps)} intervals are not 7 days " + f"(min {int(off.min())}, max {int(off.max())}). Those are not weekly " + f"changes and are not comparable to one.") + + cols = ["report_date", "long_total", "short_total", "spread_total", + "open_interest", "imbalance", "oi_gap", "balanced"] + print() + print(z.tail(args.last)[cols].to_string(index=False)) return 0 @@ -352,20 +362,17 @@ def main(argv=None) -> int: a.add_argument("--market", default=None) a.set_defaults(func=_cmd_asof) - fw = sub.add_parser("flow", help="Weekly flow decomposition per market/category.") - fw.add_argument("--market", required=True, help="CFTC market code, e.g. 088691.") - fw.add_argument("--source", choices=("vintage", "current"), default="vintage", + zs = sub.add_parser("zero-sum", + help="Check the long/short/open-interest identity per market-week.") + zs.add_argument("--market", required=True, help="CFTC market code, e.g. 088691.") + zs.add_argument("--source", choices=("vintage", "current"), default="vintage", help="'vintage' is point-in-time and starts at first capture; " "'current' reads the existing store back to 1986 but has " "revisions already applied, so it is NOT point-in-time.") - fw.add_argument("--as-of", default=None, dest="as_of", + zs.add_argument("--as-of", default=None, dest="as_of", help="Point-in-time timestamp (vintage source only).") - fw.add_argument("--category", default=None, help="e.g. noncommercial.") - fw.add_argument("--min-frac-oi", type=float, default=0.0, dest="min_frac_oi", - help="Dead zone as a fraction of prior open interest; weeks under it " - "on both legs are 'quiet'. Default 0.0 (no dead zone).") - fw.add_argument("--last", type=int, default=20, help="Rows to print (default 20).") - fw.set_defaults(func=_cmd_flow) + zs.add_argument("--last", type=int, default=20, help="Rows to print (default 20).") + zs.set_defaults(func=_cmd_zero_sum) args = p.parse_args(argv) config.store_root() # fail fast if COTDATA_STORE unset diff --git a/src/cotdata/vintage_flow.py b/src/cotdata/vintage_flow.py index af7d17b..7a3f8b8 100644 --- a/src/cotdata/vintage_flow.py +++ b/src/cotdata/vintage_flow.py @@ -1,27 +1,38 @@ -"""Flow decomposition over canonical COT rows, for crowdmon-futures module spec §6.4. - -Weekly ΔLong versus ΔShort per market/category, labelled into the four states that look -identical on a price chart and are entirely different setups: - - ΔLong +, ΔShort ~0 new longs fresh conviction buying - ΔLong ~0, ΔShort − short covering a rally with a FINITE fuel supply - ΔLong ~0, ΔShort + new shorts fresh bearish conviction - ΔLong −, ΔShort ~0 long liquidation position exit, not fresh selling - -Runs on ingested COT alone: no prices, no contract master, no multiplier. That is what -makes it the right first consumer of the canonical schema: every input is a column the -vintage layer already stores, so a failure here is a schema failure and nothing else. - -**Why this lives in cotdata.** It is a READ-side function over cotdata's own canonical -rows: it writes no store domain, adds no manifest entry, and changes no producer/consumer -contract. ADR-0007 narrows cotdata along the axis of instrument domain (COT versus bars), -not derived-versus-raw, and ADR-0008 settles that provenance for COT data stays inside -that boundary. This is the same shape of thing: COT in, COT out, nothing else consulted. -The positioning ENGINE of the module spec (extremity, fragility, cross-market PCA) is a -different matter, because it needs prices, a contract master and configured weights, and -belongs in the crowdmon package where the spec puts it. - -See docs/design/crowdmon_futures_cot_module.md §6.4. +"""The zero-sum identity over canonical COT rows, and the two ways to source them. + +`zero_sum_check` is the strongest check the canonical schema admits and it needs no +external data: in a futures market every long is somebody's short, so a single misrouted, +dropped or duplicated category column breaks the identity on the first week. That is a +statement about cotdata's own parse, which is why it lives here. + +**Flow decomposition was here and has been removed.** It was built 2026-07-30 alongside +this check, and `crowdmon.futures.flow` was built independently on 2026-08-01. The two +were carried side by side as "defensible alternatives answering slightly different +questions" until somebody measured them, which is the whole argument for measuring: + + cotdata.vintage_flow.decompose == crowdmon.futures.flow.decompose + at tolerance=1.0, gap rule off + +On 27 markets and 135,835 transitions from 2006 to 2026 that held at **100.000000% label +agreement with zero mismatches**, and `d_long`, `d_short` and `d_net` were identical on +every row. Not a similar approach. The same function, with this copy hard-wired to the +corner of the parameter space where nothing is ever `mixed` and no interval is ever +refused. Both take the dominant leg as `argmax(|ΔLong|, |ΔShort|)`, both break exact ties +to the long leg, and both treat a doubly-unmoved week as `quiet` unconditionally. + +So this was never a second opinion, it was a less capable copy: it could not decline to +call a genuinely two-sided week, and it differenced across a 294-day absence as though it +were a week. The general implementation stays, in the package the module spec puts the +positioning engine in, and is reached with `from crowdmon.futures import decompose`. + +**The dedup could not go the other way.** `crowdmon/tests/test_boundaries.py` forbids +cotdata from importing crowdmon, because a producer that depends on its consumer inverts +the whole dependency direction. The equivalence is therefore asserted from that side, in +`crowdmon/tests/test_flow_equivalence.py`, and the measurement is written up in +`crowdmon/docs/design/amendments-2026-08-02.md` §B29. + +What remains here is COT in, COT out, consulting nothing else, which is what ADR-0007 and +ADR-0008 leave inside cotdata's boundary. """ from __future__ import annotations @@ -29,149 +40,11 @@ from . import config, vintage_ingest -# The grouping key: one series per market/category, walked in report_date order. -SERIES_KEY = ["market_code", "report_type", "combined", "category"] - -NEW_LONGS = "new_longs" -SHORT_COVERING = "short_covering" -NEW_SHORTS = "new_shorts" -LONG_LIQUIDATION = "long_liquidation" -QUIET = "quiet" -FLOW_STATES = (NEW_LONGS, SHORT_COVERING, NEW_SHORTS, LONG_LIQUIDATION, QUIET) - -OUT_COLUMNS = SERIES_KEY + [ - "report_date", "days_elapsed", - "long_contracts", "short_contracts", "open_interest", - "d_long", "d_short", "d_net", "d_oi", - "state", "oi_corroborates", -] - class FlowError(ValueError): """The input is not a clean one-row-per-key-per-date panel.""" -def _require_panel(canonical: pd.DataFrame) -> None: - missing = [c for c in SERIES_KEY + ["report_date", "long_contracts", "short_contracts"] - if c not in canonical.columns] - if missing: - raise FlowError(f"missing columns for flow decomposition: {missing}") - dup = canonical.duplicated(subset=SERIES_KEY + ["report_date"]) - if dup.any(): - sample = canonical.loc[dup, SERIES_KEY + ["report_date"]].head(3).to_dict("records") - raise FlowError( - f"{int(dup.sum())} duplicate (key, report_date) rows, e.g. {sample}. " - f"A diff over a key with two rows for one week silently compares two VINTAGES " - f"of the same week and calls the revision a flow. Pass a point-in-time slice " - f"(vintage_ingest.asof) rather than raw observations.") - - -def decompose(canonical: pd.DataFrame, *, min_frac_oi: float = 0.0) -> pd.DataFrame: - """Label each week's positioning change per market/category. - - ``canonical`` is the long-form schema (``vintage_ingest.ALL_COLUMNS``), one row per - natural key per report date. Raises rather than guessing if a key has two rows for one - date, because that means two vintages and a diff across them is a revision, not a flow. - - **Classification is by dominant leg**, which is how the spec's "~0" resolves against - real data where both legs always move a little: whichever of |ΔLong|, |ΔShort| is - larger names the state, and its sign picks the direction. Exact ties go to the long - leg, so the result is deterministic. This is parameter-free by default and therefore - reproducible with nothing to tune. - - ``min_frac_oi`` optionally adds a dead zone: a week where BOTH legs move less than - that fraction of the prior week's open interest is labelled ``quiet`` instead of being - forced into a direction by noise. It defaults to 0.0 (no dead zone) deliberately: any - non-zero value is a judgement, in the same class as the module spec's fragility - weights, and belongs in a config the caller owns and can run sensitivity over, not - baked in here as a fake default. - - The first observation of each series has no predecessor and is dropped: there is no - such thing as its weekly change. - """ - _require_panel(canonical) - df = canonical.copy() - df["report_date"] = pd.to_datetime(df["report_date"]) - df = df.sort_values(SERIES_KEY + ["report_date"], kind="mergesort") - - g = df.groupby(SERIES_KEY, dropna=False, sort=False) - for src, dst in (("long_contracts", "d_long"), ("short_contracts", "d_short")): - df[dst] = g[src].diff() - df["d_oi"] = g["open_interest"].diff() if "open_interest" in df.columns else pd.NA - df["d_net"] = df["d_long"] - df["d_short"] - # Elapsed calendar days, not "one week". COT weeks are NOT uniformly seven days apart: - # holiday shifts move them, and a capture gap or an early-history hole makes a single - # diff span months. A caller comparing flow magnitudes across weeks must be able to - # see that, so it is a column rather than an assumption. - df["days_elapsed"] = g["report_date"].diff().dt.days - # Prior-week OI is the dead-zone denominator: a threshold has to be knowable BEFORE - # the week it judges, or it is fitted to the outcome it is classifying. - prior_oi = g["open_interest"].shift(1) if "open_interest" in df.columns else None - - df = df[df["d_long"].notna() & df["d_short"].notna()].copy() - df["state"] = _classify(df, prior_oi=None if prior_oi is None else prior_oi.loc[df.index], - min_frac_oi=min_frac_oi) - df["oi_corroborates"] = _corroborate(df) - return df.reindex(columns=OUT_COLUMNS).reset_index(drop=True) - - -def _classify(df: pd.DataFrame, *, prior_oi, min_frac_oi: float) -> pd.Series: - d_long = df["d_long"].astype("float64") - d_short = df["d_short"].astype("float64") - mag_l, mag_s = d_long.abs(), d_short.abs() - - long_dominates = mag_l >= mag_s # ties -> long leg, so the label is deterministic - state = pd.Series(pd.NA, index=df.index, dtype="object") - state = state.mask(long_dominates & (d_long > 0), NEW_LONGS) - state = state.mask(long_dominates & (d_long <= 0), LONG_LIQUIDATION) - state = state.mask(~long_dominates & (d_short > 0), NEW_SHORTS) - state = state.mask(~long_dominates & (d_short <= 0), SHORT_COVERING) - - # A week where NEITHER leg moved is `quiet` unconditionally, with no threshold - # involved. Without this it falls through `long_dominates & (d_long <= 0)` and is - # labelled long_liquidation, which is not a judgement call gone wrong but a plain - # misstatement: nothing was liquidated. Measured over the real 2026 Legacy file it is - # 3,308 of 29,787 transitions (11.1%), and it made long_liquidation the modal state - # with 36% of that bucket being weeks where nothing happened. The CLI prints exactly - # that value_counts() as its headline. - # - # Deliberately NOT folded into min_frac_oi. Zero is not a small number, it is the - # absence of a change, so recognising it needs no parameter and must not be switched - # off by leaving the dead zone at its default. - state = state.mask((d_long == 0) & (d_short == 0), QUIET) - - if min_frac_oi and prior_oi is not None: - dead = prior_oi.astype("float64") * float(min_frac_oi) - state = state.mask((mag_l <= dead) & (mag_s <= dead), QUIET) - return state - - -def _corroborate(df: pd.DataFrame) -> pd.Series: - """Does market open interest agree with the label? - - Futures are a closed zero-sum system, so contracts only exist because someone opened - them. Fresh positioning (new longs, new shorts) should therefore coincide with RISING - open interest, and exits (short covering, long liquidation) with falling open interest. - When it does not, the label is describing a transfer of an existing position between - categories rather than new or closed risk, which is a materially different event. - - Note the asymmetry this cannot escape: ``open_interest`` in the canonical schema is - the MARKET total, repeated on every category row, because that is what the CFTC file - reports. So this corroborates a per-category label against a market-level quantity. It - is a real check and not a proof, which is why it is a separate column rather than - something folded into the state. - """ - if "d_oi" not in df.columns: - return pd.Series(pd.NA, index=df.index, dtype="object") - d_oi = pd.to_numeric(df["d_oi"], errors="coerce") - opening = df["state"].isin([NEW_LONGS, NEW_SHORTS]) - closing = df["state"].isin([SHORT_COVERING, LONG_LIQUIDATION]) - out = pd.Series(pd.NA, index=df.index, dtype="object") - out = out.mask(opening & d_oi.notna(), d_oi > 0) - out = out.mask(closing & d_oi.notna(), d_oi < 0) - return out - - # ── Schema smoke test ─────────────────────────────────────────────────────── def zero_sum_check(canonical: pd.DataFrame) -> pd.DataFrame: """Per (report_date, market): do the category rows reconcile against open interest? @@ -238,9 +111,14 @@ def zero_sum_check(canonical: pd.DataFrame) -> pd.DataFrame: def from_vintage(*, as_of=None, market_code=None, report_type="legacy") -> pd.DataFrame: """Canonical rows out of the vintage store, as a point-in-time slice. - This is the correct source: ``asof`` returns exactly one row per natural key, so the - diff below can never straddle two vintages of one week. With no ``as_of`` it means - "everything known now", which is still a consistent single vintage per key. + This is the correct source: ``asof`` returns exactly one row per natural key, so a + consumer differencing these rows can never straddle two vintages of one week and call + a revision a flow. With no ``as_of`` it means "everything known now", which is still a + consistent single vintage per key. + + That guarantee used to be enforced here, by a `_require_panel` check in `decompose`. + It moved out with `decompose`; `crowdmon.futures.io.require_one_row_per_key` is the + same refusal on the consuming side, and this function is what makes it satisfiable. """ t = pd.Timestamp.max if as_of is None else as_of return vintage_ingest.asof(t, market_code=market_code, report_type=report_type) diff --git a/src/cotdata/vintage_ingest.py b/src/cotdata/vintage_ingest.py index a5c41e8..47c5eb5 100644 --- a/src/cotdata/vintage_ingest.py +++ b/src/cotdata/vintage_ingest.py @@ -404,7 +404,8 @@ def validate(canonical: pd.DataFrame) -> list[str]: raise ValidationError(f"categories {bad} outside vocabulary for {rt!r}") # Duplicate natural keys within ONE frame. The read side already refuses these - # (vintage_flow._require_panel); the write side must too, and for a worse reason. + # (crowdmon.futures.io.require_one_row_per_key, which is where the check went when + # vintage_flow.decompose was removed); the write side must too, and for a worse reason. # Ingesting both writes two rows sharing an identical (observed_at, snapshot_id), which # exhausts _latest_by_key's tie-break and leaves the winner decided by append order. # Both also diff against the same prior row, so revisions/ gains two contradictory diff --git a/tests/test_vintage_flow.py b/tests/test_vintage_flow.py index 92afa38..e79bed3 100644 --- a/tests/test_vintage_flow.py +++ b/tests/test_vintage_flow.py @@ -1,8 +1,28 @@ -"""Flow decomposition (module spec §6.4) and the canonical-schema smoke test. +"""The canonical-schema zero-sum smoke test. Offline and synthetic, matching the repo's test idiom. The real-data sweep that these encode is recorded in docs/design/cot_vintage.md §7: 95 markets, 149,412 weeks, every one balanced. + +**Flow decomposition used to be tested here and has moved.** It was measurably the same +function as `crowdmon.futures.flow.decompose` (100.000000% label agreement on 135,835 +transitions, at `tolerance=1.0` with the gap rule off), so the copy went rather than the +tests. Every behaviour those tests pinned is pinned on the consuming side: + +| was tested here | now | +|---|---| +| the four states by dominant leg | `crowdmon/tests/test_flow.py::test_each_pure_state` | +| first observation of a series is dropped | `..::test_first_observation_of_a_series_is_dropped_not_labelled` | +| series do not leak across market or `combined` | `..::test_deltas_never_cross_{a_market,the_combined}_boundary` | +| duplicate vintages for one week are refused | `..::test_two_vintages_of_one_week_are_refused` | +| a flat week is `quiet`, not `long_liquidation` | `..::test_neither_leg_moving_is_quiet_not_liquidation` | +| open interest corroborates or contradicts | `..::test_open_interest_corroborates_or_contradicts_the_label` | +| the equivalence itself | `crowdmon/tests/test_flow_equivalence.py` | + +The one behaviour NOT carried over is `min_frac_oi`, the optional dead zone. It defaulted +to 0.0 (off), nothing set it, and the consuming implementation resolves the same problem +with a dominance `tolerance` whose sensitivity is swept and reported. It is a deliberate +drop, not an oversight. """ import pandas as pd import pytest @@ -21,76 +41,6 @@ def _rows(dates, *, category="noncommercial", longs=(), shorts=(), oi=None, WEEKS = ["2026-01-06", "2026-01-13", "2026-01-20", "2026-01-27", "2026-02-03"] -def test_the_four_states_are_labelled_by_the_dominant_leg(): - """The spec's table is written with a '~0' leg, which never happens in real data. - Dominant-leg resolution is what makes it decidable, and it must pick the right one.""" - from cotdata import vintage_flow as vf - # start 100/100, then: +5000 long | -4000 short | +6000 short | -3000 long - can = _rows(WEEKS, - longs=[100, 5100, 5100, 5100, 2100], - shorts=[100, 100, -3900, 2100, 2100]) - fl = vf.decompose(can) - assert list(fl["state"]) == [vf.NEW_LONGS, vf.SHORT_COVERING, - vf.NEW_SHORTS, vf.LONG_LIQUIDATION] - - -def test_first_observation_of_a_series_is_dropped(): - """There is no such thing as the weekly change of the first week ever seen.""" - from cotdata import vintage_flow as vf - fl = vf.decompose(_rows(WEEKS, longs=[1, 2, 3, 4, 5], shorts=[1, 1, 1, 1, 1])) - assert len(fl) == len(WEEKS) - 1 - assert fl["report_date"].min() == pd.Timestamp(WEEKS[1]) - - -def test_series_do_not_leak_across_markets_or_categories(): - """A diff that crossed a group boundary would invent an enormous fake flow.""" - from cotdata import vintage_flow as vf - a = _rows(WEEKS[:2], longs=[10, 20], shorts=[10, 10], market="088691") - b = _rows(WEEKS[:2], longs=[9000, 9010], shorts=[10, 10], market="001602") - fl = vf.decompose(pd.concat([a, b], ignore_index=True)) - assert set(fl["d_long"]) == {10.0} # not 8980, which is the cross-market diff - - -def test_duplicate_vintages_for_one_week_are_refused(): - """Two rows for one (key, date) means two VINTAGES. Diffing them would report a - revision as though it were a flow, which is the exact confusion the vintage layer - exists to prevent.""" - from cotdata import vintage_flow as vf - can = pd.concat([_rows(WEEKS[:2], longs=[10, 20], shorts=[10, 10]), - _rows(WEEKS[1:2], longs=[25], shorts=[10])], ignore_index=True) - with pytest.raises(vf.FlowError, match="duplicate"): - vf.decompose(can) - - -def test_days_elapsed_exposes_a_gap_rather_than_assuming_seven(): - """COT was FORTNIGHTLY until 1992-10-13, and holiday weeks shift the rest. A caller - comparing flow magnitudes must be able to see that a 'weekly' change spans 15 days.""" - from cotdata import vintage_flow as vf - can = _rows(["1992-01-15", "1992-01-31", "1992-02-14"], - longs=[100, 200, 300], shorts=[100, 100, 100]) - fl = vf.decompose(can) - assert list(fl["days_elapsed"]) == [16, 14] - - -def test_dead_zone_is_off_by_default_and_scales_with_prior_open_interest(): - from cotdata import vintage_flow as vf - can = _rows(WEEKS[:2], longs=[1000, 1005], shorts=[1000, 1000], oi=[100_000, 100_000]) - assert vf.decompose(can)["state"].iloc[0] == vf.NEW_LONGS # 5 contracts, but no - # dead zone by default - quiet = vf.decompose(can, min_frac_oi=0.001) # 0.1% of 100k = 100 - assert quiet["state"].iloc[0] == vf.QUIET - - -def test_open_interest_corroborates_or_contradicts_the_label(): - """New positioning should create contracts; exits should destroy them. When it does - not, the label describes a transfer between categories, not new or closed risk.""" - from cotdata import vintage_flow as vf - rising = _rows(WEEKS[:2], longs=[100, 5100], shorts=[100, 100], oi=[1000, 9000]) - falling = _rows(WEEKS[:2], longs=[100, 5100], shorts=[100, 100], oi=[9000, 1000]) - assert vf.decompose(rising)["oi_corroborates"].iloc[0] is True - assert vf.decompose(falling)["oi_corroborates"].iloc[0] is False - - def test_zero_sum_identity_holds_and_the_gap_is_spreading(): """Every long is somebody's short, so the side totals must match. They fall short of open interest by the non-commercial SPREADING column, which providers/cftc.py does @@ -121,24 +71,17 @@ def test_missing_columns_raise_rather_than_silently_producing_nothing(): from cotdata import vintage_flow as vf can = _rows(WEEKS[:2], longs=[1, 2], shorts=[1, 1]).drop(columns=["short_contracts"]) with pytest.raises(vf.FlowError, match="missing columns"): - vf.decompose(can) + vf.zero_sum_check(can) -def test_a_week_where_nothing_moved_is_quiet_not_long_liquidation(): - """Found by adversarial review. `long_dominates & (d_long <= 0)` swallows d_long == 0, - so every flat week was labelled long_liquidation. Measured over the real 2026 Legacy - file that was 3,308 of 29,787 transitions (11.1%), which made long_liquidation the - modal state with a third of its bucket being weeks where nothing happened. - - Zero is not a small number, it is the absence of a change, so this must NOT depend on - the min_frac_oi dead zone, which is off by default.""" - from cotdata import vintage_flow as vf - can = _rows(WEEKS[:3], longs=[100, 100, 100], shorts=[50, 50, 50]) - assert list(vf.decompose(can)["state"]) == [vf.QUIET, vf.QUIET] - +def test_decompose_is_gone_and_stays_gone(): + """A regression guard against re-adding it here out of habit. -def test_a_one_contract_move_is_still_classified(): - """The quiet rule must catch only exact zero, or it becomes an unstated threshold.""" + It was re-derived once already: `crowdmon.futures.flow` was built on 2026-08-01 in + apparent ignorance that this module had held the same classifier since 2026-07-30. + The next session to want a flow decomposition should import the one that exists rather + than write a third. + """ from cotdata import vintage_flow as vf - can = _rows(WEEKS[:2], longs=[100, 101], shorts=[50, 50]) - assert vf.decompose(can)["state"].iloc[0] == vf.NEW_LONGS + assert not hasattr(vf, "decompose") + assert not hasattr(vf, "FLOW_STATES")