CFTC Weekly Swaps Report

curl 'https://www.cftc.gov/MarketReports/SwapsReports/L1GrossExpCS.html'

"""CFTC gross-notional cleared-status table -> pandas -> Parquet."""
from html.parser import HTMLParser
import requests
import pandas as pd

class FirstTable(HTMLParser):
    def __init__(self):
        super().__init__()
        self.in_table = self.done = self.in_cell = False
        self.row, self.rows, self.cell = [], [], []
    def handle_starttag(self, tag, attrs):
        if tag == "table" and not self.in_table and not self.done:
            self.in_table = True
        elif self.in_table and tag == "tr":
            self.row = []
        elif self.in_table and tag in ("td", "th"):
            self.in_cell, self.cell = True, []
    def handle_data(self, data):
        if self.in_cell:
            self.cell.append(data)
    def handle_endtag(self, tag):
        if self.in_table and tag in ("td", "th"):
            self.row.append(" ".join("".join(self.cell).split()))
            self.in_cell = False
        elif self.in_table and tag == "tr" and self.row:
            self.rows.append(self.row)
        elif self.in_table and tag == "table":
            self.in_table, self.done = False, True

URL = "https://www.cftc.gov/MarketReports/SwapsReports/L1GrossExpCS.html"
r = requests.get(URL, headers={"User-Agent": "Mozilla/5.0 SignalLabCheck/1.0"}, timeout=60)
r.raise_for_status()
p = FirstTable()
p.feed(r.text)
width = len(p.rows[0])
rows = [row for row in p.rows[1:] if len(row) == width and row[0]]
df = pd.DataFrame(rows, columns=["category", *p.rows[0][1:]])
for col in df.columns[1:]:
    df[col] = pd.to_numeric(df[col].str.replace(",", ""))
df.to_parquet("cftc_swaps_gross_notional.parquet", index=False)
print(len(df), "rows", len(df.columns) - 1, "weekly dates", "latest", df.columns[-1])