The code main.py, the file the app installs.
python Copy """CFTC Commitments of Traders: weekly positioning as a feed per futures market.
Before a run it downloads the CFTC's Commitments of Traders (COT) history for
the futures market behind the run's instrument (EURUSD reads the CME Euro FX
contract, XAUUSD the COMEX gold contract, US500 the E-mini S&P 500) and writes
one feed per market, `cot_<market>`, that a strategy reads:
def on_market(self, ctx):
bars = ctx.bars(1)
if not bars:
return
cot = ctx.data("cot_eur", bars[-1].ts)
if cot is not None and cot["cot_index"] is not None and cot["cot_index"] > 90:
ctx.signal("EXIT")
Each value is a small object: net speculator position, commercials, open
interest, the week's change, a 0 to 100 COT index over the last `index_weeks`
reports, and the trader groups of the newer reports (leveraged funds and asset
managers for financial futures, managed money for commodities).
When a value is known. A report describes positions as of Tuesday and the
CFTC publishes it on Friday at 15:30 New York time: three business days later,
pushed back one day for each federal holiday or government closure in between.
Each report is stored under the first weekday after its publication, so
Friday's report is Monday's value and a report published on a Monday after a
holiday week is Tuesday's. During the government shutdowns of 2013, 2018 to
2019 and 2025 the reports came out weeks late; the dates the CFTC announced
are in DELAYED_RELEASES and win over the rule.
The kit does the network (HTTPS to publicreporting.cftc.gov only), the atomic
write and the fill; this file is what is particular to the CFTC.
"""
import re
from datetime import date, timedelta
from orionquant_engine.plugins.kit import dates, feeds, net
HOSTS = ("publicreporting.cftc.gov",)
RESOURCE = "https://publicreporting.cftc.gov/resource/{}.json"
# The three futures only reports on the CFTC's public reporting site (Socrata
# dataset ids, checked against publicreporting.cftc.gov/api/views/<id>.json).
LEGACY = "6dca-aqww" # Legacy, Futures Only: every market, back to 1986
TFF = "gpe5-46if" # Traders in Financial Futures, Futures Only: from June 2006
DISAGGREGATED = "72hh-3qpy" # Disaggregated, Futures Only: from June 2006
#: Reports before this are left out. The CFTC moved publication to the third
#: business day after the as of date in 1992 (sixth before that, from 1990, and
#: monthly before 1990), so only from 1993 on does the rule below hold.
FIRST_REPORT = "1993-01-01"
#: A market's whole history fits one request (about 1,700 weekly rows at most).
ROW_LIMIT = 5000
#: Days a report is carried forward. Long enough to bridge the 2025 shutdown
#: (52 days between two known reports), short enough that a market the CFTC
#: stopped reporting does not look alive for months.
MAX_GAP_DAYS = 70
FEED_PREFIX = "cot_"
DELAY_RULE = "release: as of date + 3 business days (federal closures skipped), known the next weekday"
# The terms link is listed in docs.json, so the template check knows it is never fetched.
TERMS = "https://www.cftc.gov/WebPolicy/index.htm"
ATTRIBUTION = "Source: U.S. Commodity Futures Trading Commission, Commitments of Traders"
#: market key: (CFTC contract market code, trader group report, what it is).
#: Codes checked against the Legacy report on 2026-10-13; a code stays with a
#: contract when the exchange renames it.
MARKETS = {
"eur": ("099741", TFF, "Euro FX, CME"),
"jpy": ("097741", TFF, "Japanese yen, CME"),
"gbp": ("096742", TFF, "British pound, CME"),
"chf": ("092741", TFF, "Swiss franc, CME"),
"cad": ("090741", TFF, "Canadian dollar, CME"),
"aud": ("232741", TFF, "Australian dollar, CME"),
"nzd": ("112741", TFF, "New Zealand dollar, CME"),
"mxn": ("095741", TFF, "Mexican peso, CME"),
"usd_index": ("098662", TFF, "US dollar index, ICE Futures U.S."),
"sp500": ("13874A", TFF, "E-mini S&P 500, CME"),
"nasdaq100": ("209742", TFF, "E-mini Nasdaq 100, CME"),
"dow": ("124603", TFF, "DJIA x 5 dollars, CBOT"),
"russell2000": ("239742", TFF, "E-mini Russell 2000, CME"),
"nikkei": ("240741", TFF, "Nikkei Stock Average in US dollars, CME"),
"ust10y": ("043602", TFF, "10 year US Treasury note, CBOT"),
"bitcoin": ("133741", TFF, "Bitcoin, CME"),
"ether": ("146021", TFF, "Ether, CME"),
"gold": ("088691", DISAGGREGATED, "Gold, COMEX"),
"silver": ("084691", DISAGGREGATED, "Silver, COMEX"),
"copper": ("085692", DISAGGREGATED, "Copper, COMEX"),
"platinum": ("076651", DISAGGREGATED, "Platinum, NYMEX"),
"wti": ("067651", DISAGGREGATED, "WTI crude oil, NYMEX"),
"natgas": ("023651", DISAGGREGATED, "Henry Hub natural gas, NYMEX"),
}
#: Currency legs of an FX pair that have a CME contract. The contract is long
#: the currency against the US dollar, so for USDJPY the yen contract reads the
#: other way round: speculators long yen are short USDJPY.
CURRENCY_MARKETS = {
"EUR": "eur", "JPY": "jpy", "GBP": "gbp", "CHF": "chf",
"CAD": "cad", "AUD": "aud", "NZD": "nzd", "MXN": "mxn",
}
#: Instrument codes (letters and digits only, upper case) to a market. The
#: engine's own codes and their vendor spellings first, then common names.
INSTRUMENT_MARKETS = {
"XAUUSD": "gold", "GOLD": "gold",
"XAGUSD": "silver", "SILVER": "silver",
"COPPERCMDUSD": "copper", "COPPER": "copper", "XCUUSD": "copper",
"XPTUSD": "platinum", "PLATINUM": "platinum",
"LIGHTCMDUSD": "wti", "WTIUSD": "wti", "USOIL": "wti", "XTIUSD": "wti", "WTI": "wti",
"GASCMDUSD": "natgas", "NATGAS": "natgas", "XNGUSD": "natgas",
"US500": "sp500", "USA500IDXUSD": "sp500", "SPX500": "sp500", "SP500": "sp500",
"USTEC": "nasdaq100", "USATECHIDXUSD": "nasdaq100", "NAS100": "nasdaq100", "US100": "nasdaq100",
"US30": "dow", "USA30IDXUSD": "dow", "DJ30": "dow", "WS30": "dow",
"US2000": "russell2000", "USSC2000IDXUSD": "russell2000", "RUSSELL2000": "russell2000",
"JP225": "nikkei", "JPNIDXJPY": "nikkei", "JPN225": "nikkei",
"BTCUSD": "bitcoin", "BTCUSDT": "bitcoin", "XBTUSD": "bitcoin",
"ETHUSD": "ether", "ETHUSDT": "ether",
"DXY": "usd_index", "USDX": "usd_index",
"UST10Y": "ust10y", "TNOTE": "ust10y",
}
#: Reports published late because of a lapse in federal appropriations:
#: as of date to the day it was published, with the CFTC notice it comes from.
#: Where the CFTC gave only a week, the last day it could have been is used, so
#: a value is never early. The table may miss a delay nobody announced.
DELAYED_RELEASES = {
# 2013 shutdown, 1 to 16 October. CFTC press release 6745-13 (23 Oct 2013):
# the first delayed report on 25 October, the rest on a rolling basis; the
# CFTC update of 6 November 2013 says the reports through 29 October are out.
"2013-10-01": "2013-10-25",
"2013-10-08": "2013-11-06",
"2013-10-15": "2013-11-06",
"2013-10-22": "2013-11-06",
"2013-10-29": "2013-11-06",
# 2018 to 2019 shutdown, 22 December to 25 January. CFTC press release
# 7864-19 (29 Jan 2019): the 24 December report on Friday 1 February, then
# one report each Tuesday and Friday until current (8 March 2019).
"2018-12-24": "2019-02-01",
"2018-12-31": "2019-02-05",
"2019-01-08": "2019-02-08",
"2019-01-15": "2019-02-12",
"2019-01-22": "2019-02-15",
"2019-01-29": "2019-02-19",
"2019-02-05": "2019-02-22",
"2019-02-12": "2019-02-26",
"2019-02-19": "2019-03-01",
"2019-02-26": "2019-03-05",
# 2025 shutdown, 1 October to 12 November. CFTC special announcement of
# 9 December 2025 (press release "CFTC to Accelerate Publication of
# Backlogged COT Data"); the 23 December report was on time again.
"2025-09-30": "2025-11-19",
"2025-10-07": "2025-11-21",
"2025-10-14": "2025-11-25",
"2025-10-21": "2025-12-02",
"2025-10-28": "2025-12-05",
"2025-11-04": "2025-12-09",
"2025-11-10": "2025-12-10",
"2025-11-18": "2025-12-12",
"2025-11-25": "2025-12-15",
"2025-12-02": "2025-12-17",
"2025-12-09": "2025-12-19",
"2025-12-16": "2025-12-23",
}
#: Days the federal government closed besides the federal holidays, each by an
#: executive order in the Federal Register (number given) or an OPM closure.
#: Half day closures count as closed: that can only make a value later.
FEDERAL_CLOSURES = (
"1994-04-27", # EO 12910, President Nixon's funeral
"1997-12-26", # EO 13068
"1998-12-24", # EO 13109, half day
"2001-12-24", # EO 13238
"2002-12-24", # EO 13281, half day
"2003-12-26", # EO 13320
"2004-06-11", # EO 13343, President Reagan's funeral
"2007-01-02", # EO 13421, President Ford's funeral
"2007-12-24", # EO 13453
"2008-12-26", # EO 13482
"2009-12-24", # EO 13523, half day
"2012-10-29", # OPM, Hurricane Sandy
"2012-10-30", # OPM, Hurricane Sandy
"2012-12-24", # EO 13633
"2014-12-26", # EO 13682
"2015-12-24", # EO 13713, half day
"2018-12-05", # EO 13852, President George H. W. Bush's funeral
"2018-12-24", # EO 13854
"2019-12-24", # EO 13900
"2020-12-24", # EO 13965
"2024-12-24", # EO 14129
"2025-01-09", # EO 14133, President Carter's funeral
"2025-12-24", # EO 14371
"2025-12-26", # EO 14371
)
_API = {"api": None}
_CLOSED = {}
def _log(message):
if _API["api"] is not None:
_API["api"].log(message)
def _config(key, default):
api = _API["api"]
return api.get_config(key, default) if api is not None else default
# --- when a report was published -------------------------------------------------
def federal_holidays(year):
"""The days federal offices kept the holidays of 5 U.S.C. 6103 in `year`:
a Saturday holiday on the Friday before, a Sunday one on the Monday after
(so New Year's Day of the next year can land on 31 December)."""
y = int(year)
out = {
dates.observed(date(y, 1, 1)),
dates.observed(date(y + 1, 1, 1)),
dates.nth_weekday(y, 1, dates.MON, 3), # Martin Luther King Jr. Day, since 1986
dates.nth_weekday(y, 2, dates.MON, 3), # Washington's Birthday
dates.nth_weekday(y, 5, dates.MON, -1), # Memorial Day
dates.observed(date(y, 7, 4)),
dates.nth_weekday(y, 9, dates.MON, 1), # Labor Day
dates.nth_weekday(y, 10, dates.MON, 2), # Columbus Day
dates.observed(date(y, 11, 11)), # Veterans Day
dates.nth_weekday(y, 11, dates.THU, 4), # Thanksgiving
dates.observed(date(y, 12, 25)),
}
if y >= 2021:
out.add(dates.observed(date(y, 6, 19))) # Juneteenth, a federal holiday from 17 June 2021
return {d for d in out if d.year == y}
def closed_days(year):
"""Federal holidays and closures in `year`, cached per year."""
y = int(year)
if y not in _CLOSED:
extra = {date.fromisoformat(d) for d in FEDERAL_CLOSURES if d.startswith(str(y))}
_CLOSED[y] = frozenset(federal_holidays(y) | extra)
return _CLOSED[y]
def release_day(report):
"""The day the CFTC published the report with this as of date.
The table of delayed releases first; otherwise three business days after
the as of date, skipping federal holidays and closures. That is the rule
the CFTC states (published on the third business day, one day later per
holiday) and it reproduces its published 2026 schedule.
"""
day = dates.as_day(report)
listed = DELAYED_RELEASES.get(day.isoformat())
if listed:
return date.fromisoformat(listed)
holidays = closed_days(day.year) | closed_days(day.year + 1)
return dates.add_business_days(day, 3, holidays=holidays)
def known_day(report):
"""The first day a strategy may use a report: the weekday after it was
published (15:30 New York time, after most of that day's bars)."""
day = release_day(report) + timedelta(days=1)
while day.weekday() > dates.FRI:
day += timedelta(days=1)
return day
# --- which markets ------------------------------------------------------------------
def markets_for_instrument(instrument):
"""The market keys behind an engine instrument code, [] when none.
An FX pair maps to the CME contract of each non dollar leg that has one
(EURUSD to eur, USDJPY to jpy, EURJPY to eur and jpy); anything else is
looked up in INSTRUMENT_MARKETS under any spelling (US500, USA500.IDX/USD).
"""
flat = re.sub(r"[^A-Z0-9]", "", str(instrument or "").upper())
if flat in INSTRUMENT_MARKETS:
return [INSTRUMENT_MARKETS[flat]]
if len(flat) == 6 and flat.isalpha():
return [CURRENCY_MARKETS[leg] for leg in (flat[:3], flat[3:]) if leg in CURRENCY_MARKETS]
return []
def wanted_markets(setting, instrument):
"""Market keys from the `markets` setting: a comma separated list of keys,
where `auto` stands for the run's instrument. Unknown names are logged."""
out = []
for token in re.split(r"[,\s]+", str(setting or "").strip().lower()):
if not token:
continue
if token == "auto":
found = markets_for_instrument(instrument)
if not found:
_log("no CFTC market for {}: no COT feed for this run (set markets to one of {})".format(
instrument, ", ".join(sorted(MARKETS))))
out += found
elif token in MARKETS:
out.append(token)
else:
_log("{!r} is not a market this plugin knows: {}".format(token, ", ".join(sorted(MARKETS))))
return list(dict.fromkeys(out))
# --- the download -------------------------------------------------------------------
def _query(dataset, code, columns):
params = {
"$select": ",".join(["report_date_as_yyyy_mm_dd"] + list(columns)),
"$where": "cftc_contract_market_code='{}' AND report_date_as_yyyy_mm_dd >= '{}'".format(code, FIRST_REPORT),
"$order": "report_date_as_yyyy_mm_dd",
"$limit": ROW_LIMIT,
}
rows = net.get_json(RESOURCE.format(dataset), hosts=HOSTS, params=params)
if not isinstance(rows, list):
raise net.NetError("network", "the CFTC answered with something that is not a list of rows")
if len(rows) >= ROW_LIMIT:
_log("the CFTC returned {} rows for {}, the limit: the oldest may be missing".format(len(rows), code))
return rows
def _number(row, key):
raw = row.get(key)
if raw in (None, ""):
return None
try:
return int(round(float(raw)))
except (TypeError, ValueError):
return None
def _net(row, long_key, short_key):
long_, short = _number(row, long_key), _number(row, short_key)
if long_ is None or short is None:
return None
return long_ - short
def _day(row):
text = str(row.get("report_date_as_yyyy_mm_dd") or "")[:10]
try:
return date.fromisoformat(text)
except ValueError:
return None
LEGACY_COLUMNS = (
"open_interest_all",
"noncomm_positions_long_all", "noncomm_positions_short_all",
"change_in_noncomm_long_all", "change_in_noncomm_short_all",
"comm_positions_long_all", "comm_positions_short_all",
)
TFF_COLUMNS = (
"lev_money_positions_long", "lev_money_positions_short",
"asset_mgr_positions_long", "asset_mgr_positions_short",
)
DISAGGREGATED_COLUMNS = ("m_money_positions_long_all", "m_money_positions_short_all")
def parse_legacy(rows):
"""Legacy rows to {as of date: fields}, oldest first."""
out = {}
for row in rows:
day = _day(row)
spec = _net(row, "noncomm_positions_long_all", "noncomm_positions_short_all")
if day is None or spec is None:
continue
oi = _number(row, "open_interest_all")
out[day] = {
"report_date": day.isoformat(),
"open_interest": oi,
"net_speculators": spec,
"net_speculators_change": _net(row, "change_in_noncomm_long_all", "change_in_noncomm_short_all"),
"net_speculators_pct_oi": round(100.0 * spec / oi, 2) if oi else None,
"net_commercials": _net(row, "comm_positions_long_all", "comm_positions_short_all"),
}
return {d: out[d] for d in sorted(out)}
def parse_groups(rows, dataset):
"""Trader group rows (TFF or Disaggregated) to {as of date: fields}."""
out = {}
for row in rows:
day = _day(row)
if day is None:
continue
if dataset == TFF:
fields = {
"net_leveraged_funds": _net(row, "lev_money_positions_long", "lev_money_positions_short"),
"net_asset_managers": _net(row, "asset_mgr_positions_long", "asset_mgr_positions_short"),
}
else:
fields = {"net_managed_money": _net(row, "m_money_positions_long_all", "m_money_positions_short_all")}
out[day] = fields
return out
def cot_index(values, weeks):
"""Where each value sits in the range of the last `weeks` values, ending
with itself, from 0 (the lowest) to 100 (the highest). None until there are
`weeks` values, and when the range is flat. Only past values are used."""
n = max(2, int(weeks))
out = []
for i in range(len(values)):
window = [v for v in values[max(0, i - n + 1): i + 1] if v is not None]
if i + 1 < n or len(window) < n or values[i] is None:
out.append(None)
continue
low, high = min(window), max(window)
out.append(None if high == low else round(100.0 * (values[i] - low) / (high - low), 1))
return out
def build_reports(legacy, groups, group_names, weeks):
"""One value per as of date: the Legacy fields, the trader groups of the
same week (None when that report has no such week), and the COT index."""
days = list(legacy)
index = cot_index([legacy[d]["net_speculators"] for d in days], weeks)
reports = {}
for d, idx in zip(days, index, strict=True):
value = dict(legacy[d])
value["cot_index"] = idx
extra = groups.get(d) or {}
for name in group_names:
value[name] = extra.get(name)
value["released"] = release_day(d).isoformat()
reports[d] = value
return reports
def fetch_market(key, weeks):
"""Download one market and write its feed. Returns True when written."""
code, dataset, label = MARKETS[key]
group_columns = TFF_COLUMNS if dataset == TFF else DISAGGREGATED_COLUMNS
group_names = (("net_leveraged_funds", "net_asset_managers") if dataset == TFF else ("net_managed_money",))
legacy = parse_legacy(_query(LEGACY, code, LEGACY_COLUMNS))
if not legacy:
_log("the CFTC returned no reports for {} ({}), the cached feed stays".format(key, code))
return False
groups = parse_groups(_query(dataset, code, group_columns), dataset)
reports = build_reports(legacy, groups, group_names, weeks)
known = feeds.shift(reports, known_day)
feeds.write_feed(
FEED_PREFIX + key,
feeds.fill_forward(known, until=feeds.today(), max_gap_days=MAX_GAP_DAYS),
meta={
"source": "CFTC Commitments of Traders, futures only: {} (code {})".format(label, code),
"terms": TERMS,
"attribution": ATTRIBUTION,
"delay_rule": DELAY_RULE,
"market": key,
"contract_code": code,
"datasets": [LEGACY, dataset],
"index_weeks": int(weeks),
"delayed_releases": len(DELAYED_RELEASES),
},
)
_log("cached {} COT reports for {} ({})".format(len(reports), key, label))
return True
def data_fetch(context):
"""Lifecycle hook: refresh the COT feeds the run needs before it starts."""
instrument = getattr(context, "instrument", "")
keys = wanted_markets(_config("markets", "auto"), instrument)
if not keys:
return
weeks = int(float(_config("index_weeks", 156)))
if weeks < 2:
_log("index_weeks {} is too small, 2 is the least".format(weeks))
weeks = 2
max_age = float(_config("cache_days", 1)) * 86400
for key in keys:
name = FEED_PREFIX + key
meta = feeds.read_meta(name) or {}
if meta.get("index_weeks") == weeks and feeds.fresh(name, max_age):
continue
try:
fetch_market(key, weeks)
except net.NetError as exc:
_log("fetch failed for {}, the cached feed stays: {}".format(key, exc))
def register(api):
_API["api"] = api
api.register_data_fetch(data_fetch)