deemanone/beanflows
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add CFTC COT data integration with foundation data model layer

Browse Source 
- New extraction package (cftc_cot): downloads yearly Disaggregated Futures ZIPs
  from CFTC, etag-based dedup, dynamic inner filename discovery, gzip normalization
- SQLMesh 3-layer architecture: raw (technical) → foundation (business model) → serving (mart)
- dim_commodity seed: conformed dimension mapping USDA ↔ CFTC codes — the commodity ontology
- fct_cot_positioning: typed, deduplicated weekly positioning facts for all commodities
- obt_cot_positioning: Coffee C mart with COT Index (26w/52w), WoW delta, OI ratios
- Analytics functions + REST API endpoints: /commodities/<code>/positioning[/latest]
- Dashboard widget: Managed Money net, COT Index card, dual-axis Chart.js chart
- 23 passing tests (10 unit + 2 SQLMesh model + existing regression suite)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
2026-02-20 21:57:04 +01:00
parent d09ba91023
commit 0a83b2cb74
19 changed files with 1111 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
transform/sqlmesh_materia/macros/__init__.py
View File

@@ -8,3 +8,10 @@ def psd_glob(evaluator) -> str:
"""Return a quoted glob path for all PSD CSV gzip files under LANDING_DIR."""
landing_dir = evaluator.var("LANDING_DIR") or os.environ.get("LANDING_DIR", "data/landing")
return f"'{landing_dir}/psd/**/*.csv.gzip'"
@macro()
def cot_glob(evaluator) -> str:
"""Return a quoted glob path for all COT CSV gzip files under LANDING_DIR."""
landing_dir = evaluator.var("LANDING_DIR") or os.environ.get("LANDING_DIR", "data/landing")
return f"'{landing_dir}/cot/**/*.csv.gzip'"

24
transform/sqlmesh_materia/models/foundation/dim_commodity.sql Normal file
View File

@@ -0,0 +1,24 @@
-- Commodity dimension: conforms identifiers across source systems.
--
-- This is the ontology seed. Each row is a commodity tracked by BeanFlows.
-- As new sources are added (ICO, futures prices, satellite), their
-- commodity identifiers are added as columns here — not as separate tables.
-- As new commodities are added (cocoa, sugar), rows are added here.
--
-- References:
-- usda_commodity_code → raw.psd_alldata.commodity_code
-- cftc_commodity_code → raw.cot_disaggregated.cftc_commodity_code
MODEL (
name foundation.dim_commodity,
kind SEED (
path '$root/seeds/dim_commodity.csv',
csv_settings (delimiter = ';')
),
columns (
usda_commodity_code varchar,
cftc_commodity_code varchar,
commodity_name varchar,
commodity_group varchar
)
);

160
transform/sqlmesh_materia/models/foundation/fct_cot_positioning.sql Normal file
View File

@@ -0,0 +1,160 @@
-- Foundation fact: CFTC COT positioning, weekly grain, all commodities.
--
-- Casts raw varchar columns to proper types, cleans column names,
-- computes net positions (long - short) per trader category, and
-- deduplicates via hash key. Covers all commodities — filtering to
-- a specific commodity happens in the serving layer.
--
-- Grain: one row per (cftc_commodity_code, report_date, cftc_contract_market_code)
-- History: revisions appear as new rows with a later ingest_date.
-- Serving layer picks max(ingest_date) per grain for latest view.
MODEL (
name foundation.fct_cot_positioning,
kind INCREMENTAL_BY_TIME_RANGE (
time_column report_date
),
grain (cftc_commodity_code, report_date, cftc_contract_market_code, ingest_date),
start '2006-06-13',
cron '@daily'
);
WITH cast_and_clean AS (
SELECT
-- Identifiers
trim(market_and_exchange_names) AS market_and_exchange_name,
report_date_as_yyyy_mm_dd::date AS report_date,
trim(cftc_commodity_code) AS cftc_commodity_code,
trim(cftc_contract_market_code) AS cftc_contract_market_code,
trim(contract_units) AS contract_units,
-- Open interest
open_interest_all::int AS open_interest,
-- Producer / Merchant (commercial hedgers: exporters, processors)
prod_merc_positions_long_all::int AS prod_merc_long,
prod_merc_positions_short_all::int AS prod_merc_short,
-- Swap dealers
swap_positions_long_all::int AS swap_long,
swap_positions_short_all::int AS swap_short,
swap_positions_spread_all::int AS swap_spread,
-- Managed money (hedge funds, CTAs — the primary speculative signal)
m_money_positions_long_all::int AS managed_money_long,
m_money_positions_short_all::int AS managed_money_short,
m_money_positions_spread_all::int AS managed_money_spread,
-- Other reportables
other_rept_positions_long_all::int AS other_reportable_long,
other_rept_positions_short_all::int AS other_reportable_short,
other_rept_positions_spread_all::int AS other_reportable_spread,
-- Non-reportable (small speculators, below reporting threshold)
nonrept_positions_long_all::int AS nonreportable_long,
nonrept_positions_short_all::int AS nonreportable_short,
-- Net positions (long minus short per category)
prod_merc_positions_long_all::int
- prod_merc_positions_short_all::int AS prod_merc_net,
m_money_positions_long_all::int
- m_money_positions_short_all::int AS managed_money_net,
swap_positions_long_all::int
- swap_positions_short_all::int AS swap_net,
other_rept_positions_long_all::int
- other_rept_positions_short_all::int AS other_reportable_net,
nonrept_positions_long_all::int
- nonrept_positions_short_all::int AS nonreportable_net,
-- Week-over-week changes
change_in_open_interest_all::int AS change_open_interest,
change_in_m_money_long_all::int AS change_managed_money_long,
change_in_m_money_short_all::int AS change_managed_money_short,
change_in_m_money_long_all::int
- change_in_m_money_short_all::int AS change_managed_money_net,
change_in_prod_merc_long_all::int AS change_prod_merc_long,
change_in_prod_merc_short_all::int AS change_prod_merc_short,
-- Concentration ratios (% of OI held by top 4 / top 8 traders)
conc_gross_le_4_tdr_long_all::float AS concentration_top4_long_pct,
conc_gross_le_4_tdr_short_all::float AS concentration_top4_short_pct,
conc_gross_le_8_tdr_long_all::float AS concentration_top8_long_pct,
conc_gross_le_8_tdr_short_all::float AS concentration_top8_short_pct,
-- Trader counts
traders_tot_all::int AS traders_total,
traders_m_money_long_all::int AS traders_managed_money_long,
traders_m_money_short_all::int AS traders_managed_money_short,
traders_m_money_spread_all::int AS traders_managed_money_spread,
-- Ingest date: derived from landing path year directory
-- Path: .../cot/{year}/{etag}.csv.gzip → extract year from [-2]
make_date(split(filename, '/')[-2]::int, 1, 1) AS ingest_date,
-- Dedup key: hash of business grain + key metrics
hash(
cftc_commodity_code,
report_date_as_yyyy_mm_dd,
cftc_contract_market_code,
open_interest_all,
m_money_positions_long_all,
m_money_positions_short_all,
prod_merc_positions_long_all,
prod_merc_positions_short_all
) AS hkey
FROM raw.cot_disaggregated
-- Reject rows with null commodity code or malformed date
WHERE trim(cftc_commodity_code) IS NOT NULL
AND len(trim(cftc_commodity_code)) > 0
AND report_date_as_yyyy_mm_dd::date IS NOT NULL
),
deduplicated AS (
SELECT
any_value(market_and_exchange_name) AS market_and_exchange_name,
any_value(report_date) AS report_date,
any_value(cftc_commodity_code) AS cftc_commodity_code,
any_value(cftc_contract_market_code) AS cftc_contract_market_code,
any_value(contract_units) AS contract_units,
any_value(open_interest) AS open_interest,
any_value(prod_merc_long) AS prod_merc_long,
any_value(prod_merc_short) AS prod_merc_short,
any_value(prod_merc_net) AS prod_merc_net,
any_value(swap_long) AS swap_long,
any_value(swap_short) AS swap_short,
any_value(swap_spread) AS swap_spread,
any_value(swap_net) AS swap_net,
any_value(managed_money_long) AS managed_money_long,
any_value(managed_money_short) AS managed_money_short,
any_value(managed_money_spread) AS managed_money_spread,
any_value(managed_money_net) AS managed_money_net,
any_value(other_reportable_long) AS other_reportable_long,
any_value(other_reportable_short) AS other_reportable_short,
any_value(other_reportable_spread) AS other_reportable_spread,
any_value(other_reportable_net) AS other_reportable_net,
any_value(nonreportable_long) AS nonreportable_long,
any_value(nonreportable_short) AS nonreportable_short,
any_value(nonreportable_net) AS nonreportable_net,
any_value(change_open_interest) AS change_open_interest,
any_value(change_managed_money_long) AS change_managed_money_long,
any_value(change_managed_money_short) AS change_managed_money_short,
any_value(change_managed_money_net) AS change_managed_money_net,
any_value(change_prod_merc_long) AS change_prod_merc_long,
any_value(change_prod_merc_short) AS change_prod_merc_short,
any_value(concentration_top4_long_pct) AS concentration_top4_long_pct,
any_value(concentration_top4_short_pct) AS concentration_top4_short_pct,
any_value(concentration_top8_long_pct) AS concentration_top8_long_pct,
any_value(concentration_top8_short_pct) AS concentration_top8_short_pct,
any_value(traders_total) AS traders_total,
any_value(traders_managed_money_long) AS traders_managed_money_long,
any_value(traders_managed_money_short) AS traders_managed_money_short,
any_value(traders_managed_money_spread) AS traders_managed_money_spread,
any_value(ingest_date) AS ingest_date,
hkey
FROM cast_and_clean
GROUP BY hkey
)
SELECT *
FROM deduplicated
WHERE report_date BETWEEN @start_ds AND @end_ds

85
transform/sqlmesh_materia/models/raw/cot_disaggregated.sql Normal file
View File

@@ -0,0 +1,85 @@
-- Raw CFTC Commitment of Traders — Disaggregated Futures Only.
--
-- Technical ingestion layer only: reads gzip CSVs from the landing directory
-- and surfaces the columns needed by downstream foundation models.
-- All values are varchar; casting happens in foundation.
--
-- Source: CFTC yearly ZIPs at
-- https://www.cftc.gov/files/dea/history/fut_disagg_txt_{year}.zip
-- Coverage: June 2006 – present (new file every Friday at 3:30 PM ET)
MODEL (
name raw.cot_disaggregated,
kind FULL,
grain (cftc_commodity_code, report_date_as_yyyy_mm_dd, cftc_contract_market_code),
start '2006-06-13',
cron '@daily'
);
SELECT
-- Identifiers
"Market_and_Exchange_Names" AS market_and_exchange_names,
"Report_Date_as_YYYY-MM-DD" AS report_date_as_yyyy_mm_dd,
"CFTC_Commodity_Code" AS cftc_commodity_code,
"CFTC_Contract_Market_Code" AS cftc_contract_market_code,
"Contract_Units" AS contract_units,
-- Open interest
"Open_Interest_All" AS open_interest_all,
-- Producer / Merchant / Processor / User (commercial hedgers)
"Prod_Merc_Positions_Long_All" AS prod_merc_positions_long_all,
"Prod_Merc_Positions_Short_All" AS prod_merc_positions_short_all,
-- Swap dealers
"Swap_Positions_Long_All" AS swap_positions_long_all,
"Swap__Positions_Short_All" AS swap_positions_short_all,
"Swap__Positions_Spread_All" AS swap_positions_spread_all,
-- Managed money (hedge funds, CTAs — key speculative signal)
"M_Money_Positions_Long_All" AS m_money_positions_long_all,
"M_Money_Positions_Short_All" AS m_money_positions_short_all,
"M_Money_Positions_Spread_All" AS m_money_positions_spread_all,
-- Other reportables
"Other_Rept_Positions_Long_All" AS other_rept_positions_long_all,
"Other_Rept_Positions_Short_All" AS other_rept_positions_short_all,
"Other_Rept_Positions_Spread_All" AS other_rept_positions_spread_all,
-- Non-reportable (small speculators)
"NonRept_Positions_Long_All" AS nonrept_positions_long_all,
"NonRept_Positions_Short_All" AS nonrept_positions_short_all,
-- Week-over-week changes
"Change_in_Open_Interest_All" AS change_in_open_interest_all,
"Change_in_M_Money_Long_All" AS change_in_m_money_long_all,
"Change_in_M_Money_Short_All" AS change_in_m_money_short_all,
"Change_in_Prod_Merc_Long_All" AS change_in_prod_merc_long_all,
"Change_in_Prod_Merc_Short_All" AS change_in_prod_merc_short_all,
-- Concentration (% of OI held by top 4 and top 8 traders)
"Conc_Gross_LE_4_TDR_Long_All" AS conc_gross_le_4_tdr_long_all,
"Conc_Gross_LE_4_TDR_Short_All" AS conc_gross_le_4_tdr_short_all,
"Conc_Gross_LE_8_TDR_Long_All" AS conc_gross_le_8_tdr_long_all,
"Conc_Gross_LE_8_TDR_Short_All" AS conc_gross_le_8_tdr_short_all,
-- Trader counts
"Traders_Tot_All" AS traders_tot_all,
"Traders_M_Money_Long_All" AS traders_m_money_long_all,
"Traders_M_Money_Short_All" AS traders_m_money_short_all,
"Traders_M_Money_Spread_All" AS traders_m_money_spread_all,
-- Lineage
filename
FROM read_csv(
@cot_glob(),
delim = ',',
encoding = 'utf-8',
compression = 'gzip',
header = true,
union_by_name = true,
filename = true,
all_varchar = true,
max_line_size = 10000000,
ignore_errors = true
)

140
transform/sqlmesh_materia/models/serving/obt_cot_positioning.sql Normal file
View File

@@ -0,0 +1,140 @@
-- Serving mart: COT positioning for Coffee C futures, analytics-ready.
--
-- Joins foundation.fct_cot_positioning with foundation.dim_commodity so
-- the coffee filter is driven by the dimension (not a hardcoded CFTC code).
-- Adds derived analytics used by the dashboard and API:
-- - Normalized positioning (% of open interest)
-- - Long/short ratio
-- - Week-over-week momentum
-- - COT Index over 26-week and 52-week trailing windows (0=bearish, 100=bullish)
--
-- Grain: one row per report_date for Coffee C futures.
-- Latest revision per date: MAX(ingest_date) used to deduplicate CFTC corrections.
MODEL (
name serving.cot_positioning,
kind INCREMENTAL_BY_TIME_RANGE (
time_column report_date
),
grain (report_date),
start '2006-06-13',
cron '@daily'
);
WITH latest_revision AS (
-- Pick the most recently ingested row when CFTC issues corrections
SELECT f.*
FROM foundation.fct_cot_positioning f
INNER JOIN foundation.dim_commodity d
ON f.cftc_commodity_code = d.cftc_commodity_code
WHERE d.commodity_name = 'Coffee, Green'
AND f.report_date BETWEEN @start_ds AND @end_ds
QUALIFY ROW_NUMBER() OVER (
PARTITION BY f.report_date, f.cftc_contract_market_code
ORDER BY f.ingest_date DESC
) = 1
),
with_derived AS (
SELECT
report_date,
market_and_exchange_name,
cftc_commodity_code,
cftc_contract_market_code,
contract_units,
ingest_date,
-- Absolute positions (contracts)
open_interest,
managed_money_long,
managed_money_short,
managed_money_spread,
managed_money_net,
prod_merc_long,
prod_merc_short,
prod_merc_net,
swap_long,
swap_short,
swap_spread,
swap_net,
other_reportable_long,
other_reportable_short,
other_reportable_spread,
other_reportable_net,
nonreportable_long,
nonreportable_short,
nonreportable_net,
-- Normalized: managed money net as % of open interest
-- Removes size effects and makes cross-period comparison meaningful
round(
managed_money_net::float / NULLIF(open_interest, 0) * 100,
2
) AS managed_money_net_pct_of_oi,
-- Long/short ratio: >1 = more bulls than bears in managed money
round(
managed_money_long::float / NULLIF(managed_money_short, 0),
3
) AS managed_money_long_short_ratio,
-- Weekly changes
change_open_interest,
change_managed_money_long,
change_managed_money_short,
change_managed_money_net,
change_prod_merc_long,
change_prod_merc_short,
-- Week-over-week momentum in managed money net (via LAG)
managed_money_net - LAG(managed_money_net, 1) OVER (
ORDER BY report_date
) AS managed_money_net_wow,
-- Concentration
concentration_top4_long_pct,
concentration_top4_short_pct,
concentration_top8_long_pct,
concentration_top8_short_pct,
-- Trader counts
traders_total,
traders_managed_money_long,
traders_managed_money_short,
traders_managed_money_spread,
-- COT Index (26-week): where is current net vs. trailing 26 weeks?
-- 0 = most bearish extreme, 100 = most bullish extreme
-- Industry-standard sentiment gauge (equivalent to RSI for positioning)
CASE
WHEN MAX(managed_money_net) OVER w26 = MIN(managed_money_net) OVER w26
THEN 50.0
ELSE round(
(managed_money_net - MIN(managed_money_net) OVER w26)::float
/ (MAX(managed_money_net) OVER w26 - MIN(managed_money_net) OVER w26)
* 100,
1
)
END AS cot_index_26w,
-- COT Index (52-week): longer-term positioning context
CASE
WHEN MAX(managed_money_net) OVER w52 = MIN(managed_money_net) OVER w52
THEN 50.0
ELSE round(
(managed_money_net - MIN(managed_money_net) OVER w52)::float
/ (MAX(managed_money_net) OVER w52 - MIN(managed_money_net) OVER w52)
* 100,
1
)
END AS cot_index_52w
FROM latest_revision
WINDOW
w26 AS (ORDER BY report_date ROWS BETWEEN 25 PRECEDING AND CURRENT ROW),
w52 AS (ORDER BY report_date ROWS BETWEEN 51 PRECEDING AND CURRENT ROW)
)
SELECT *
FROM with_derived
ORDER BY report_date

2
transform/sqlmesh_materia/seeds/dim_commodity.csv Normal file
View File

@@ -0,0 +1,2 @@
usda_commodity_code;cftc_commodity_code;commodity_name;commodity_group
0711100;083731;Coffee, Green;Softs
1 usda_commodity_code cftc_commodity_code commodity_name commodity_group
2 0711100 083731 Coffee, Green Softs

99
transform/sqlmesh_materia/tests/test_cot_foundation.yaml Normal file
View File

@@ -0,0 +1,99 @@
test_fct_cot_positioning_types_and_net_positions:
model: foundation.fct_cot_positioning
inputs:
raw.cot_disaggregated:
rows:
- market_and_exchange_names: "COFFEE C - ICE FUTURES U.S."
report_date_as_yyyy_mm_dd: "2024-01-02"
cftc_commodity_code: "083731"
cftc_contract_market_code: "083731"
contract_units: "37,500 POUNDS"
open_interest_all: "250000"
prod_merc_positions_long_all: "80000"
prod_merc_positions_short_all: "90000"
swap_positions_long_all: "30000"
swap_positions_short_all: "35000"
swap_positions_spread_all: "10000"
m_money_positions_long_all: "60000"
m_money_positions_short_all: "40000"
m_money_positions_spread_all: "15000"
other_rept_positions_long_all: "20000"
other_rept_positions_short_all: "18000"
other_rept_positions_spread_all: "5000"
nonrept_positions_long_all: "12000"
nonrept_positions_short_all: "14000"
change_in_open_interest_all: "5000"
change_in_m_money_long_all: "2000"
change_in_m_money_short_all: "-1000"
change_in_prod_merc_long_all: "1000"
change_in_prod_merc_short_all: "500"
conc_gross_le_4_tdr_long_all: "35.5"
conc_gross_le_4_tdr_short_all: "28.3"
conc_gross_le_8_tdr_long_all: "52.1"
conc_gross_le_8_tdr_short_all: "44.7"
traders_tot_all: "450"
traders_m_money_long_all: "85"
traders_m_money_short_all: "62"
traders_m_money_spread_all: "20"
filename: "data/landing/cot/2024/abc123.csv.gzip"
expected:
rows:
- report_date: "2024-01-02"
cftc_commodity_code: "083731"
open_interest: 250000
managed_money_long: 60000
managed_money_short: 40000
managed_money_net: 20000
prod_merc_long: 80000
prod_merc_short: 90000
prod_merc_net: -10000
swap_long: 30000
swap_short: 35000
swap_net: -5000
nonreportable_long: 12000
nonreportable_short: 14000
nonreportable_net: -2000
change_managed_money_net: 3000
traders_managed_money_long: 85
traders_managed_money_short: 62
test_fct_cot_positioning_rejects_null_commodity:
model: foundation.fct_cot_positioning
inputs:
raw.cot_disaggregated:
rows:
- market_and_exchange_names: "SOME OTHER CONTRACT"
report_date_as_yyyy_mm_dd: "2024-01-02"
cftc_commodity_code: ""
cftc_contract_market_code: "999999"
contract_units: "N/A"
open_interest_all: "1000"
prod_merc_positions_long_all: "500"
prod_merc_positions_short_all: "500"
swap_positions_long_all: "0"
swap_positions_short_all: "0"
swap_positions_spread_all: "0"
m_money_positions_long_all: "0"
m_money_positions_short_all: "0"
m_money_positions_spread_all: "0"
other_rept_positions_long_all: "0"
other_rept_positions_short_all: "0"
other_rept_positions_spread_all: "0"
nonrept_positions_long_all: "0"
nonrept_positions_short_all: "0"
change_in_open_interest_all: "0"
change_in_m_money_long_all: "0"
change_in_m_money_short_all: "0"
change_in_prod_merc_long_all: "0"
change_in_prod_merc_short_all: "0"
conc_gross_le_4_tdr_long_all: "0"
conc_gross_le_4_tdr_short_all: "0"
conc_gross_le_8_tdr_long_all: "0"
conc_gross_le_8_tdr_short_all: "0"
traders_tot_all: "10"
traders_m_money_long_all: "0"
traders_m_money_short_all: "0"
traders_m_money_spread_all: "0"
filename: "data/landing/cot/2024/abc123.csv.gzip"
expected:
rows: []