Commitments of traders (CoT) primer analysis

This is a primer on the analysis of the Commitments of Traders (CoT) report. The goal here is just to showcase how to add this weekly report, into a wider analytical framework of your choice. I’ve also provided commented code, so you can reproduce the results locally.

For the sake of brevity, I will not expand on what this report covers, the assumption here is that you already have some understanding. For an explanation please refer to any decent source on the web, i.e. Wikipedia, etc.

In this post I will focus on the EUR/USD currency pair in 2023 report. There’s no particular reason why, it’s just a popular pair with lots publicly available of data, both on the spot and futures market.

A popular way of using the CoT as an indicator, is to take an asset from the futures market and plot it against the spot asset, i.e. EUR/USD futures price from the CoT report will be drawn against the EUR/USD weekly spot. This could help to spot extreme positions in either large speculators or large commercials, and could, depending on the market, signal trend continuation or reversal.

1. Data Link to heading

  • Ticker: EUR/USD

  • Type:

    • Spot market: OHLCV
    • Futures market: Open Interest

  • Timeframe: calendar year 2023

  • Frequency: weekly


First point of order is retrieving this report:


# cot_downloader.py
#
# Download all available CoT FuturesOnly reports from CFTC website

import requests

from bs4 import BeautifulSoup
from pathlib import Path
from zipfile import ZipFile


def cot_bulk_downloader():
    """
    Downloads all standard Commitment of Traders Futures only reports.

    :return: 0 if ok | 1 if error -> int
    """
    url = 'https://www.cftc.gov/MarketReports/CommitmentsofTraders/HistoricalCompressed/index.htm'
    out_path = Path.cwd() / 'datasets' / 'commitments-of-traders'
    # avoid re-downloading if the files have already been downloaded
    if out_path.exists():
        return 0
    Path.mkdir(out_path, exist_ok=True, parents=True)
    try:
        response = requests.get(url)
        soup = BeautifulSoup(response.text, 'html.parser')
        for name in soup.findAll('a', href=True):
            file_url = name['href']
            if file_url.endswith('.zip'):
                # split based on url "/" path (all files are .zip) and retain "filename.zip"
                filename = file_url.split('/')[-1]
                if 'deacot' in filename:
                    out_name = out_path.joinpath(filename)
                    zip_url = 'https://www.cftc.gov/files/dea/history/' + filename
                    # the GET request triggers the download action
                    resp = requests.get(zip_url, stream=True)
                    if resp.status_code == requests.codes.ok:
                        print(f'Downloading {filename} . . .')
                        with open(out_name, 'wb') as f:
                            for chunk in resp.iter_content():
                                if chunk:
                                    f.write(chunk)
                            f.close()
                        print('Finished downloading report')
    except Exception as error:
        print(f"Unexpected error: {error}")
        return 1

    return 0


def bulk_unzipper(target_dir):
    """
    Unzips all archives within the input directory.

    :param target_dir: input directory -> Path-like object
    :return: 0 if ok | 1 if error -> int
    """

    # if the directory exists and there's a zip file in it,
    # then proceed with bulk unzipping
    try:
        for file in target_dir.iterdir():
            if str(file).endswith('.zip'):
                print(f'Unzipping {file} . . .')
                with ZipFile(file, 'r') as zip_object:
                    filename = zip_object.filename.split('/')[-1]  # after split consider only the 'deacotXXXX.zip part
                    filename = filename.split('.zip')[0]           # consider only the 'deacotXXXX' string
                    zip_object.extractall(target_dir / filename)
    except Exception as error:
        print(f"Unexpected error: {error}")
        return 1

    return 0


def cot_data_bootstrap():
    """
    Seed other modules with data.

    :return: int -> 0 if ok, 1 if error
    """
    try:
        return_val = cot_bulk_downloader()
        # all is well so proceed to unzip
        if return_val == 0:
            # path below is created by bulk downloader so no I/O error
            cot_dir = Path.cwd() / 'datasets' / 'commitments-of-traders'
            bulk_unzipper(cot_dir)
    except Exception as error:
        print(f'Unexpected error: {error}')
        return 1

    return 0

2. The Analysis Link to heading

Now that the futures data has been downloaded, we can start thinking of what to do with it. Since this is a primer, I will compute relatively easy metrics,

  • net positions of large speculators and commercials (aka hedgers)
  • market sentiment
  • speculators sentiment

where,

$ net\ position = long\ contracts\ - short\ contracts $

$ market\ sentiment = speculators\ net\ position\ - commercials\ net\ position $.

$ speculators\ sentiment = \dfrac{speculators\ long\ contracts}{total\ speculators\ contracts} * 100 $ (%)


I made up the last formula, but I find it sensible to have a ratio to quickly gauge if speculators are heavily long/short in relative terms, so if speculator sentiment is for example 35%, then large traders are heavily short. That could be noise or could be something worth looking into – it depends on how intimate is one’s knowledge in any given market.

The code for the formulas above is:


# cot_yearly_analysis.py

def net_positions(instrument):
    """
    Computes the net position for large speculators and hedgers, where,

    net_position = long contracts - short contracts

    e.g., speculator net position = speculator long contracts - speculators short contracts.

    Non-reportable positions are omitted (aka small speculator positions)

    :param instrument: futures instrument represented as the ticker -> str
    :return: net positions of speculators and hedgers respectively -> tuple(int)
    """
    speculators_net_position = instrument[2] - instrument[3]
    commercials_net_position = instrument[5] - instrument[6]
    return speculators_net_position, commercials_net_position


def market_sentiment(instruments):
    """
    Compute market sentiment from CoT report, where,
    
    market sentiment = speculators net position - commercials net position
    
    :param instruments: iterable of instruments represented as tickers -> list
    :return: delta between net positions of large speculators and commercials -> int
    """
    non_commercials_net_position, commercials_net_position = net_positions(instruments)
    return non_commercials_net_position - commercials_net_position


def speculators_sentiment(instrument):
    """
    Compute speculator long and short positions (contracts) from CoT report, where,

    speculators_long = speculators long / (spec. long + spec. short)
    speculators_short = speculators short / (spec. long + spec. short)

    :param instrument: futures instrument represented as the ticker -> str
    :return: speculators long positions and short positions -> tuple(float)
    """
    speculators_long = instrument[2] / (instrument[2] + instrument[3])
    speculators_short = instrument[3] / (instrument[2] + instrument[3])
    return round(speculators_long, 4), round(speculators_short, 4)

Now that the formulas are out of the way, we can look at the results in the next section. But first I’ll need to show the code used to draw the plot:


# plotter.py

import plotly.graph_objects as go

from pathlib import Path
from plotly.subplots import make_subplots


def generic_time_series(df, x_column_name, y_column_name, **kwargs):
    """
    Plots a time series.

    Assumptions when plotting the time series:

    - X-axis is a Date or DateTimeIndex (sensible for a time series)
    - Y-axis is a pandas Series

    :param df: input Dataframe -> pandas Dataframe
    :param x_column_name: X-axis pandas column name -> str
    :param y_column_name: Y-axis pandas column name -> str
    :param kwargs: optional args -> dict
    :return: Null
    """
    fig = make_subplots(
        rows=2,
        cols=1,
        shared_xaxes=True,
        shared_yaxes=False
    )
    # N.B.: when using the graph objects API from Plotly, Line plots are rendered with the
    # "Scatter" class for some reason.

    # add main plot/trace
    fig.add_trace(
        go.Scatter(
            x=df.index.values,
            y=df[y_column_name],
            mode="lines",
            name=kwargs.get("y_label")
        ),
        row=1,
        col=1
    )
    # there's one or more to additional plots
    if "extra_x" or "extra_y_1" or "extra_y_2" in kwargs:
        # add secondary trace(s), aka actual subplots in Plotly lingo
        fig.add_trace(
            go.Scatter(
                x=df.index.values,
                y=kwargs.get("extra_y_1"),  # there has to be a y-axis on the secondary plot
                mode="lines",
                name=kwargs.get("extra_y_1_legend_label", "")
            ),
            row=2,
            col=1
        )

        fig.add_trace(
            go.Scatter(
                x=df.index.values,
                y=kwargs.get("extra_y_2"),  # there has to be a y-axis on the secondary plot
                mode="lines",
                name=kwargs.get("extra_y_2_legend_label", "")
            ),
            row=2,
            col=1
        )
        # update y-axis properties of sub trace(s)
        y_subplot_label = " ".join(kwargs.get("extra_y_1_legend_label").split(" ")[1:])
        fig.update_yaxes(title_text=y_subplot_label, row=2, col=1)

    # Canvas styling
    fig.update_xaxes(tickangle=45)
    fig.update_yaxes(title_text=kwargs.get("y_label", "Y"), row=1, col=1)
    fig.update_layout(title_text=kwargs.get("plot_title", "No title provided"))

    # exist_ok = True makes sure "charts" directory is not created or overwritten if it
    # already exists
    Path.cwd().joinpath("charts").mkdir(exist_ok=True)
    filename = f"charts/{kwargs.get('filename', 'Y')}"
    fig.write_image(f"{filename}.svg")
    fig.write_html(f"{filename}.html")

3. Results and concluding remarks Link to heading

Taken as is, there is no discernible pattern or connection between net positions from the CoT and the spot market for EUR/USD, yet it doesn’t mean the report is useless. After all I only looked at a year’s worth of data. Additionally, supply & demand levels – where a report like this would shine – are nearly impossible to estimate with currencies given central banks liberal money printing practices. For a physical commodity market it should be more useful, given that supply & demand are linked to physical inventories, i.e. a cattle trader cannot print a billion more cows overnight , a central bank can with the domestic currency.

To make the most of this report, it’s best to collect the measurements across several points in time, because market conditions change, e.g. net positions that seem extreme in current market conditions might not be so extreme if measured against another point in the past. Building an index like the CPI is the best way to maximise the value of the CoT report.

Finally, the driver code that brings all the code snippets together:

# driver.py

if __name__ == '__main__':
    import pandas as pd
    from cot_downloader import cot_data_bootstrap, futures_dir, Path
    from yahoo_finance_downloader import prepare_ticker_for_yf_download
    from cot_analysis import market_sentiment, net_positions, speculators_sentiment
    from plotter import generic_time_series

    cot_data_bootstrap()  # seed application with data

    # visually inspect the report .txt to find these column names
    columns = [
        'Market and Exchange Names',
        'As of Date in Form YYYY-MM-DD',
        'Noncommercial Positions-Long (All)',
        'Noncommercial Positions-Short (All)',
        'Noncommercial Positions-Spreading (All)',
        'Commercial Positions-Long (All)',
        'Commercial Positions-Short (All)',
        'Nonreportable Positions-Long (All)',
        'Nonreportable Positions-Short (All)'
    ]

    # in a more dynamic setting, these vars will come from user command line input or
    # web form, etc.
    focus_instrument = "EUR/USD"
    focus_year = "2023"
    start_date = "2022-12-31"
    end_date = "2023-12-31"
    analysis_title = "Weekly Close prices"
    currency = "USD"
    frequency = "1wk"

    futures_data = pd.read_csv(
        Path.cwd() / futures_dir / f"deacot{focus_year}" / 'annual.txt',
        delimiter=',',
        usecols=columns
    )

    spot_instrument = prepare_ticker_for_yf_download(focus_instrument)
    spot_data = yf.Ticker(spot_instrument).history(
      start=start_date, end=end_date, interval=frequency
    )

    # map spot asset to futures market equivalent. For stocks or similar, they're 
    # equivalent, but for currencies they are different, e.g "EUR/USD" spot maps to
    # "EURO FX" in the futures market.
    spot_to_futures_map = {
        "EUR/USD": "EURO FX",
        "EURUSD=X": "EURO FX",
    }
    if spot_instrument in spot_to_futures_map.keys():
        futures_instrument = spot_to_futures_map[spot_instrument]
    else:
        futures_instrument = focus_instrument

    speculators = []
    hedgers = []
    for market in futures_data.values:
        if futures_instrument in market[0]:
            market_delta = market_sentiment(market)
            speculators_net_position, hedgers_net_position = net_positions(market)
            speculators_long, speculators_short = speculators_sentiment(market)
            speculators.append(
              (market[1], speculators_net_position, speculators_long, speculators_short)
            )
            hedgers.append((market[1], hedgers_net_position))

            # primary plot with historical spot data
            x_column_name = spot_data.index.name  # _name is a column name of the Dataframe
            y_column_name = "Close"
            x_label = "Date"                      # _label is what is shown on the plot
            y_label = f"{spot_instrument} {y_column_name} price ({currency})"

            generic_time_series(
                spot_data,
                x_column_name,
                y_column_name,
                filename=f"{spot_instrument}_{data_frequency}_{focus_year}",
                plot_title=f"{spot_instrument} {analysis_title} ({currency}) {focus_year}",
                x_label=x_label,
                y_label=y_label,
                # secondary plot(s) data
                extra_y_1=[net_position[1] for net_position in speculators],
                extra_y_2=[net_position[1] for net_position in hedgers],
                extra_y_1_legend_label='Speculators net position',
                extra_y_2_legend_label='Hedgers net position',
            )

What about other commodities or currencies? Link to heading

I only covered the EUR/USD pair for convenience. If you want me to focus on other instruments, you can email me at here.

Or if you found this analysis interesting feel free to share on the social media links below.



Share on:
LinkedIn