Resources Reference

Resource client modules mapping to COT Bias API endpoints.

AsyncInferenceResource

AsyncInferenceResource(client: AsyncClient)

Asynchronous resource for querying current week AI positioning regimes and attribution.

Initialize the async inference resource with an active async HTTP client.

Parameters:

Name	Type	Description	Default
client	AsyncClient	Active asynchronous HTTP client session.	required

Source code in src/cotbias/resources/inference.py

def __init__(self, client: httpx.AsyncClient) -> None:
    """Initialize the async inference resource with an active async HTTP client.

    Args:
        client: Active asynchronous HTTP client session.
    """
    self._client = client

get async

get(symbol: str) -> InferenceSchema

Fetch current week's AI positioning regime and macro attribution asynchronously.

Parameters:

Name	Type	Description	Default
symbol	str	Asset ticker symbol (e.g., 'GC', 'ES', 'CL', '6E', 'ZB').	required

Returns:

Name	Type	Description
An	InferenceSchema	class:~cotbias.models.inference.InferenceSchema model containing
	InferenceSchema	institutional bias score, regime status, primary driver, momentum,
	InferenceSchema	crowdedness risk, and qualitative macro attribution rationale.

Raises:

Type	Description
AuthenticationError	If API key is missing or invalid (HTTP 401).
PermissionDeniedError	If account tier lacks access (HTTP 403).
NotFoundError	If the market symbol is not found (HTTP 404).
RateLimitExceededError	If daily request quota is exceeded (HTTP 429).
ServerError	If upstream COT Bias servers return an error (HTTP 5xx).
CotApiError	Base exception for any unexpected API errors.

Example

import asyncio
from cotbias import AsyncCotClient

async def main():
    async with AsyncCotClient() as client:
        regime = await client.inference.get("CL")
        print(regime.ai_bias_score, regime.momentum)

asyncio.run(main())

Source code in src/cotbias/resources/inference.py

async def get(self, symbol: str) -> InferenceSchema:
    """Fetch current week's AI positioning regime and macro attribution asynchronously.

    Args:
        symbol: Asset ticker symbol (e.g., 'GC', 'ES', 'CL', '6E', 'ZB').

    Returns:
        An :class:`~cotbias.models.inference.InferenceSchema` model containing
        institutional bias score, regime status, primary driver, momentum,
        crowdedness risk, and qualitative macro attribution rationale.

    Raises:
        AuthenticationError: If API key is missing or invalid (HTTP 401).
        PermissionDeniedError: If account tier lacks access (HTTP 403).
        NotFoundError: If the market symbol is not found (HTTP 404).
        RateLimitExceededError: If daily request quota is exceeded (HTTP 429).
        ServerError: If upstream COT Bias servers return an error (HTTP 5xx).
        CotApiError: Base exception for any unexpected API errors.

    Example:
        ```python
        import asyncio
        from cotbias import AsyncCotClient

        async def main():
            async with AsyncCotClient() as client:
                regime = await client.inference.get("CL")
                print(regime.ai_bias_score, regime.momentum)

        asyncio.run(main())
        ```
    """
    resp = await self._client.get(f"/inference/{symbol.upper()}")
    handle_error_response(resp)
    return InferenceSchema.model_validate(resp.json())

InferenceResource

InferenceResource(client: Client)

Synchronous resource for querying current week AI positioning regimes and attribution.

Initialize the inference resource with an active HTTP client.

Parameters:

Name	Type	Description	Default
client	Client	Active synchronous HTTP client session.	required

Source code in src/cotbias/resources/inference.py

def __init__(self, client: httpx.Client) -> None:
    """Initialize the inference resource with an active HTTP client.

    Args:
        client: Active synchronous HTTP client session.
    """
    self._client = client

get

get(symbol: str) -> InferenceSchema

Fetch current week's AI positioning regime and macro attribution for an asset.

Parameters:

Name	Type	Description	Default
symbol	str	Asset ticker symbol (e.g., 'GC', 'ES', 'CL', '6E', 'ZB').	required

Returns:

Name	Type	Description
An	InferenceSchema	class:~cotbias.models.inference.InferenceSchema model containing
	InferenceSchema	institutional bias score, regime status, primary driver, momentum,
	InferenceSchema	crowdedness risk, and qualitative macro attribution rationale.

Raises:

Type	Description
AuthenticationError	If API key is missing or invalid (HTTP 401).
PermissionDeniedError	If account tier lacks access (HTTP 403).
NotFoundError	If the market symbol is not found (HTTP 404).
RateLimitExceededError	If daily request quota is exceeded (HTTP 429).
ServerError	If upstream COT Bias servers return an error (HTTP 5xx).
CotApiError	Base exception for any unexpected API errors.

Example

from cotbias import CotClient

client = CotClient()
regime = client.inference.get("ES")
print(f"Bias: {regime.ai_bias_score} | Regime: {regime.regime_status}")
print(f"Rationale: {regime.qualitative_attribution.macro_rationale}")

Source code in src/cotbias/resources/inference.py

def get(self, symbol: str) -> InferenceSchema:
    """Fetch current week's AI positioning regime and macro attribution for an asset.

    Args:
        symbol: Asset ticker symbol (e.g., 'GC', 'ES', 'CL', '6E', 'ZB').

    Returns:
        An :class:`~cotbias.models.inference.InferenceSchema` model containing
        institutional bias score, regime status, primary driver, momentum,
        crowdedness risk, and qualitative macro attribution rationale.

    Raises:
        AuthenticationError: If API key is missing or invalid (HTTP 401).
        PermissionDeniedError: If account tier lacks access (HTTP 403).
        NotFoundError: If the market symbol is not found (HTTP 404).
        RateLimitExceededError: If daily request quota is exceeded (HTTP 429).
        ServerError: If upstream COT Bias servers return an error (HTTP 5xx).
        CotApiError: Base exception for any unexpected API errors.

    Example:
        ```python
        from cotbias import CotClient

        client = CotClient()
        regime = client.inference.get("ES")
        print(f"Bias: {regime.ai_bias_score} | Regime: {regime.regime_status}")
        print(f"Rationale: {regime.qualitative_attribution.macro_rationale}")
        ```
    """
    resp = self._client.get(f"/inference/{symbol.upper()}")
    handle_error_response(resp)
    return InferenceSchema.model_validate(resp.json())

AsyncMarketsResource

AsyncMarketsResource(client: AsyncClient)

Asynchronous resource for querying the catalog of tracked futures contracts.

Initialize the async markets resource with an active async HTTP client.

Parameters:

Name	Type	Description	Default
client	AsyncClient	Active asynchronous HTTP client session.	required

Source code in src/cotbias/resources/markets.py

def __init__(self, client: httpx.AsyncClient) -> None:
    """Initialize the async markets resource with an active async HTTP client.

    Args:
        client: Active asynchronous HTTP client session.
    """
    self._client = client

list async

list(sector: str | None = None) -> list[MarketMetadata]

List tracked futures contracts asynchronously.

Parameters:

Name	Type	Description	Default
sector	str | None	Optional sector filter: 'Commodity', 'Equity', 'FX', or 'Rates'. If omitted, returns all tracked contracts.	None

Returns:

Type	Description
list[MarketMetadata]	A list of :class:~cotbias.models.markets.MarketMetadata objects detailing
list[MarketMetadata]	symbol, descriptive name, CFTC commodity code, sector, and supported report types.

Raises:

Type	Description
AuthenticationError	If API key is missing or invalid (HTTP 401).
PermissionDeniedError	If account tier lacks access (HTTP 403).
RateLimitExceededError	If daily request quota is exceeded (HTTP 429).
ServerError	If upstream COT Bias servers return an error (HTTP 5xx).
CotApiError	Base exception for any unexpected API errors.

Example

import asyncio
from cotbias import AsyncCotClient

async def main():
    async with AsyncCotClient() as client:
        all_markets = await client.markets.list()
        print(f"Total contracts tracked: {len(all_markets)}")

asyncio.run(main())

Source code in src/cotbias/resources/markets.py

async def list(self, sector: str | None = None) -> list[MarketMetadata]:
    """List tracked futures contracts asynchronously.

    Args:
        sector: Optional sector filter: `'Commodity'`, `'Equity'`, `'FX'`, or `'Rates'`.
            If omitted, returns all tracked contracts.

    Returns:
        A list of :class:`~cotbias.models.markets.MarketMetadata` objects detailing
        symbol, descriptive name, CFTC commodity code, sector, and supported report types.

    Raises:
        AuthenticationError: If API key is missing or invalid (HTTP 401).
        PermissionDeniedError: If account tier lacks access (HTTP 403).
        RateLimitExceededError: If daily request quota is exceeded (HTTP 429).
        ServerError: If upstream COT Bias servers return an error (HTTP 5xx).
        CotApiError: Base exception for any unexpected API errors.

    Example:
        ```python
        import asyncio
        from cotbias import AsyncCotClient

        async def main():
            async with AsyncCotClient() as client:
                all_markets = await client.markets.list()
                print(f"Total contracts tracked: {len(all_markets)}")

        asyncio.run(main())
        ```
    """
    params = {"sector": sector} if sector else {}
    resp = await self._client.get("/markets", params=params)
    handle_error_response(resp)
    parsed = MarketsResponse.model_validate(resp.json())
    return parsed.markets

MarketsResource

MarketsResource(client: Client)

Synchronous resource for querying the catalog of tracked futures contracts.

Initialize the markets resource with an active HTTP client.

Parameters:

Name	Type	Description	Default
client	Client	Active synchronous HTTP client session.	required

Source code in src/cotbias/resources/markets.py

def __init__(self, client: httpx.Client) -> None:
    """Initialize the markets resource with an active HTTP client.

    Args:
        client: Active synchronous HTTP client session.
    """
    self._client = client

list

list(sector: str | None = None) -> list[MarketMetadata]

List tracked futures contracts with sector and CFTC report mappings.

Parameters:

Name	Type	Description	Default
sector	str | None	Optional sector filter: 'Commodity', 'Equity', 'FX', or 'Rates'. If omitted, returns all tracked contracts.	None

Returns:

Type	Description
list[MarketMetadata]	A list of :class:~cotbias.models.markets.MarketMetadata objects detailing
list[MarketMetadata]	symbol, descriptive name, CFTC commodity code, sector, and supported report types.

Raises:

Type	Description
AuthenticationError	If API key is missing or invalid (HTTP 401).
PermissionDeniedError	If account tier lacks access (HTTP 403).
RateLimitExceededError	If daily request quota is exceeded (HTTP 429).
ServerError	If upstream COT Bias servers return an error (HTTP 5xx).
CotApiError	Base exception for any unexpected API errors.

Example

from cotbias import CotClient

client = CotClient()
fx_markets = client.markets.list(sector="FX")
for m in fx_markets:
    print(f"{m.symbol} ({m.name}): CFTC #{m.cftc_code}")

Source code in src/cotbias/resources/markets.py

def list(self, sector: str | None = None) -> list[MarketMetadata]:
    """List tracked futures contracts with sector and CFTC report mappings.

    Args:
        sector: Optional sector filter: `'Commodity'`, `'Equity'`, `'FX'`, or `'Rates'`.
            If omitted, returns all tracked contracts.

    Returns:
        A list of :class:`~cotbias.models.markets.MarketMetadata` objects detailing
        symbol, descriptive name, CFTC commodity code, sector, and supported report types.

    Raises:
        AuthenticationError: If API key is missing or invalid (HTTP 401).
        PermissionDeniedError: If account tier lacks access (HTTP 403).
        RateLimitExceededError: If daily request quota is exceeded (HTTP 429).
        ServerError: If upstream COT Bias servers return an error (HTTP 5xx).
        CotApiError: Base exception for any unexpected API errors.

    Example:
        ```python
        from cotbias import CotClient

        client = CotClient()
        fx_markets = client.markets.list(sector="FX")
        for m in fx_markets:
            print(f"{m.symbol} ({m.name}): CFTC #{m.cftc_code}")
        ```
    """
    params = {"sector": sector} if sector else {}
    resp = self._client.get("/markets", params=params)
    handle_error_response(resp)
    parsed = MarketsResponse.model_validate(resp.json())
    return parsed.markets

AsyncScreenerResource

AsyncScreenerResource(client: AsyncClient)

Asynchronous resource for multi-market positioning scans.

Initialize the async screener resource with an active async HTTP client.

Parameters:

Name	Type	Description	Default
client	AsyncClient	Active asynchronous HTTP client session.	required

Source code in src/cotbias/resources/screener.py

def __init__(self, client: httpx.AsyncClient) -> None:
    """Initialize the async screener resource with an active async HTTP client.

    Args:
        client: Active asynchronous HTTP client session.
    """
    self._client = client

scan async

scan(ai_bias_score: str | None = ..., crowdedness_risk: str | None = ..., momentum: str | None = ..., cot_positioning: str | None = ..., as_dataframe: Literal[True] = ...) -> pd.DataFrame

scan(ai_bias_score: str | None = ..., crowdedness_risk: str | None = ..., momentum: str | None = ..., cot_positioning: str | None = ..., as_dataframe: Literal[False] = ...) -> ScreenerResponse

scan(ai_bias_score: str | None = ..., crowdedness_risk: str | None = ..., momentum: str | None = ..., cot_positioning: str | None = ..., as_dataframe: bool = ...) -> pd.DataFrame | ScreenerResponse

scan(ai_bias_score: str | None = None, crowdedness_risk: str | None = None, momentum: str | None = None, cot_positioning: str | None = None, as_dataframe: bool = True) -> pd.DataFrame | ScreenerResponse

Filter across all tracked futures markets asynchronously.

Parameters:

Name	Type	Description	Default
ai_bias_score	str | None	Filter by directional bias ('Bullish', 'Bearish', 'Neutral').	None
crowdedness_risk	str | None	Filter by positioning risk ('High Liquidation Risk', 'Balanced').	None
momentum	str | None	Filter by open interest dynamics ('New Longs', 'Short Covering').	None
cot_positioning	str | None	Filter by smart vs dumb positioning ('Convergence', 'Divergence').	None
as_dataframe	bool	When True (default), returns a Pandas DataFrame. When False, returns a ScreenerResponse.	True

Returns:

Type	Description
DataFrame | ScreenerResponse	A Pandas DataFrame if as_dataframe=True, or a ScreenerResponse model if as_dataframe=False.

Raises:

Type	Description
AuthenticationError	If API key is missing or invalid (HTTP 401).
PermissionDeniedError	If account tier lacks access (HTTP 403).
RateLimitExceededError	If daily request quota is exceeded (HTTP 429).
ServerError	If upstream COT Bias servers return an error (HTTP 5xx).
CotApiError	Base exception for any unexpected API errors.

Example

import asyncio
from cotbias import AsyncCotClient

async def main():
    async with AsyncCotClient() as client:
        results = await client.screener.scan(ai_bias_score="Bullish")
        print(results)

asyncio.run(main())

Source code in src/cotbias/resources/screener.py

async def scan(
    self,
    ai_bias_score: str | None = None,
    crowdedness_risk: str | None = None,
    momentum: str | None = None,
    cot_positioning: str | None = None,
    as_dataframe: bool = True,
) -> pd.DataFrame | ScreenerResponse:
    """Filter across all tracked futures markets asynchronously.

    Args:
        ai_bias_score: Filter by directional bias ('Bullish', 'Bearish', 'Neutral').
        crowdedness_risk: Filter by positioning risk ('High Liquidation Risk', 'Balanced').
        momentum: Filter by open interest dynamics ('New Longs', 'Short Covering').
        cot_positioning: Filter by smart vs dumb positioning ('Convergence', 'Divergence').
        as_dataframe: When `True` (default), returns a Pandas DataFrame.
            When `False`, returns a `ScreenerResponse`.

    Returns:
        A Pandas DataFrame if `as_dataframe=True`, or a `ScreenerResponse` model if `as_dataframe=False`.

    Raises:
        AuthenticationError: If API key is missing or invalid (HTTP 401).
        PermissionDeniedError: If account tier lacks access (HTTP 403).
        RateLimitExceededError: If daily request quota is exceeded (HTTP 429).
        ServerError: If upstream COT Bias servers return an error (HTTP 5xx).
        CotApiError: Base exception for any unexpected API errors.

    Example:
        ```python
        import asyncio
        from cotbias import AsyncCotClient

        async def main():
            async with AsyncCotClient() as client:
                results = await client.screener.scan(ai_bias_score="Bullish")
                print(results)

        asyncio.run(main())
        ```
    """
    params: dict[str, str] = {}
    if ai_bias_score:
        params["ai_bias_score"] = ai_bias_score
    if crowdedness_risk:
        params["crowdedness_risk"] = crowdedness_risk
    if momentum:
        params["momentum"] = momentum
    if cot_positioning:
        params["cot_positioning"] = cot_positioning

    resp = await self._client.get("/screener", params=params)
    handle_error_response(resp)

    parsed = ScreenerResponse.model_validate(resp.json())

    if not as_dataframe:
        return parsed

    raw_results = [r.model_dump() for r in parsed.results]
    return cast("pd.DataFrame", pd.DataFrame(raw_results))

ScreenerResource

ScreenerResource(client: Client)

Synchronous resource for multi-market positioning scans.

Initialize the screener resource with an active HTTP client.

Parameters:

Name	Type	Description	Default
client	Client	Active synchronous HTTP client session.	required

Source code in src/cotbias/resources/screener.py

def __init__(self, client: httpx.Client) -> None:
    """Initialize the screener resource with an active HTTP client.

    Args:
        client: Active synchronous HTTP client session.
    """
    self._client = client

scan

scan(ai_bias_score: str | None = ..., crowdedness_risk: str | None = ..., momentum: str | None = ..., cot_positioning: str | None = ..., as_dataframe: Literal[True] = ...) -> pd.DataFrame

scan(ai_bias_score: str | None = ..., crowdedness_risk: str | None = ..., momentum: str | None = ..., cot_positioning: str | None = ..., as_dataframe: Literal[False] = ...) -> ScreenerResponse

scan(ai_bias_score: str | None = ..., crowdedness_risk: str | None = ..., momentum: str | None = ..., cot_positioning: str | None = ..., as_dataframe: bool = ...) -> pd.DataFrame | ScreenerResponse

scan(ai_bias_score: str | None = None, crowdedness_risk: str | None = None, momentum: str | None = None, cot_positioning: str | None = None, as_dataframe: bool = True) -> pd.DataFrame | ScreenerResponse

Filter across all tracked futures markets based on positioning classifications.

Parameters:

Name	Type	Description	Default
ai_bias_score	str | None	Filter by directional bias (e.g. 'Bullish', 'Strongly Bullish', 'Bearish', 'Neutral').	None
crowdedness_risk	str | None	Filter by positioning risk (e.g. 'High Liquidation Risk', 'High Squeeze Risk', 'Balanced').	None
momentum	str | None	Filter by open interest dynamics (e.g. 'New Longs (OI Expanding)', 'Short Covering (OI Contracting)').	None
cot_positioning	str | None	Filter by smart vs dumb positioning ('Convergence' or 'Divergence').	None
as_dataframe	bool	When True (default), returns a Pandas DataFrame of matched contracts. When False, returns a :class:~cotbias.models.screener.ScreenerResponse.	True

Returns:

Type	Description
DataFrame | ScreenerResponse	A Pandas DataFrame if as_dataframe=True, or a ScreenerResponse model if as_dataframe=False.

Raises:

Type	Description
AuthenticationError	If API key is missing or invalid (HTTP 401).
PermissionDeniedError	If account tier lacks access (HTTP 403).
RateLimitExceededError	If daily request quota is exceeded (HTTP 429).
ServerError	If upstream COT Bias servers return an error (HTTP 5xx).
CotApiError	Base exception for any unexpected API errors.

Example

from cotbias import CotClient

client = CotClient()
df = client.screener.scan(
    ai_bias_score="Bullish",
    crowdedness_risk="Balanced",
    as_dataframe=True,
)
print(df[["symbol", "regime_status", "primary_driver"]])

Source code in src/cotbias/resources/screener.py

def scan(
    self,
    ai_bias_score: str | None = None,
    crowdedness_risk: str | None = None,
    momentum: str | None = None,
    cot_positioning: str | None = None,
    as_dataframe: bool = True,
) -> pd.DataFrame | ScreenerResponse:
    """Filter across all tracked futures markets based on positioning classifications.

    Args:
        ai_bias_score: Filter by directional bias (e.g. `'Bullish'`, `'Strongly Bullish'`,
            `'Bearish'`, `'Neutral'`).
        crowdedness_risk: Filter by positioning risk (e.g. `'High Liquidation Risk'`,
            `'High Squeeze Risk'`, `'Balanced'`).
        momentum: Filter by open interest dynamics (e.g. `'New Longs (OI Expanding)'`,
            `'Short Covering (OI Contracting)'`).
        cot_positioning: Filter by smart vs dumb positioning (`'Convergence'` or `'Divergence'`).
        as_dataframe: When `True` (default), returns a Pandas DataFrame of matched
            contracts. When `False`, returns a :class:`~cotbias.models.screener.ScreenerResponse`.

    Returns:
        A Pandas DataFrame if `as_dataframe=True`, or a `ScreenerResponse` model if `as_dataframe=False`.

    Raises:
        AuthenticationError: If API key is missing or invalid (HTTP 401).
        PermissionDeniedError: If account tier lacks access (HTTP 403).
        RateLimitExceededError: If daily request quota is exceeded (HTTP 429).
        ServerError: If upstream COT Bias servers return an error (HTTP 5xx).
        CotApiError: Base exception for any unexpected API errors.

    Example:
        ```python
        from cotbias import CotClient

        client = CotClient()
        df = client.screener.scan(
            ai_bias_score="Bullish",
            crowdedness_risk="Balanced",
            as_dataframe=True,
        )
        print(df[["symbol", "regime_status", "primary_driver"]])
        ```
    """
    params: dict[str, str] = {}
    if ai_bias_score:
        params["ai_bias_score"] = ai_bias_score
    if crowdedness_risk:
        params["crowdedness_risk"] = crowdedness_risk
    if momentum:
        params["momentum"] = momentum
    if cot_positioning:
        params["cot_positioning"] = cot_positioning

    resp = self._client.get("/screener", params=params)
    handle_error_response(resp)

    parsed = ScreenerResponse.model_validate(resp.json())

    if not as_dataframe:
        return parsed

    raw_results = [r.model_dump() for r in parsed.results]
    return cast("pd.DataFrame", pd.DataFrame(raw_results))

AsyncTimeseriesResource

AsyncTimeseriesResource(client: AsyncClient)

Asynchronous resource for querying historical weekly COT positioning timeseries.

Initialize the async timeseries resource with an active async HTTP client.

Parameters:

Name	Type	Description	Default
client	AsyncClient	Active asynchronous HTTP client session.	required

Source code in src/cotbias/resources/timeseries.py

def __init__(self, client: httpx.AsyncClient) -> None:
    """Initialize the async timeseries resource with an active async HTTP client.

    Args:
        client: Active asynchronous HTTP client session.
    """
    self._client = client

get async

get(symbol: str, contract_type: ContractType = ..., lookback_weeks: int = ..., report: ReportType = ..., as_dataframe: Literal[True] = ...) -> pd.DataFrame

get(symbol: str, contract_type: ContractType = ..., lookback_weeks: int = ..., report: ReportType = ..., as_dataframe: Literal[False] = ...) -> TimeseriesResponse

get(symbol: str, contract_type: str = ..., lookback_weeks: int = ..., report: str = ..., as_dataframe: bool = ...) -> pd.DataFrame | TimeseriesResponse

get(symbol: str, contract_type: str = 'futopt', lookback_weeks: int = 156, report: str = 'legacy', as_dataframe: bool = True) -> pd.DataFrame | TimeseriesResponse

Fetch historical weekly positioning metrics for a market asynchronously.

Parameters:

Name	Type	Description	Default
symbol	str	Asset ticker symbol (e.g., 'GC', 'ES', 'CL', '6E', 'ZB').	required
contract_type	str	Position breakdown type: - 'fut': Futures only. - 'futopt': Combined Futures and Options (default). - 'opt': Options delta equivalent only.	'futopt'
lookback_weeks	int	Number of weekly CFTC reports to retrieve (1 to 156, default 156).	156
report	str	CFTC report family ('legacy', 'disaggregated', 'tff', 'supplemental').	'legacy'
as_dataframe	bool	When True (default), returns a Pandas DataFrame indexed by report_date in ascending order. When False, returns a validated TimeseriesResponse model.	True

Returns:

Type	Description
DataFrame | TimeseriesResponse	A DatetimeIndex-sorted Pandas DataFrame if as_dataframe=True, or a
DataFrame | TimeseriesResponse	TimeseriesResponse Pydantic model if as_dataframe=False.

Raises:

Type	Description
AuthenticationError	If API key is missing or invalid (HTTP 401).
PermissionDeniedError	If account tier lacks access (HTTP 403).
NotFoundError	If symbol is unrecognized or report data is missing (HTTP 404).
RateLimitExceededError	If daily request quota is exceeded (HTTP 429).
ServerError	If upstream COT Bias servers return an error (HTTP 5xx).
CotApiError	Base exception for any unexpected API errors.

Example

import asyncio
from cotbias import AsyncCotClient

async def main():
    async with AsyncCotClient() as client:
        df = await client.timeseries.get("GC", lookback_weeks=26)
        print(df.tail())

asyncio.run(main())

Source code in src/cotbias/resources/timeseries.py

async def get(
    self,
    symbol: str,
    contract_type: str = "futopt",
    lookback_weeks: int = 156,
    report: str = "legacy",
    as_dataframe: bool = True,
) -> pd.DataFrame | TimeseriesResponse:
    """Fetch historical weekly positioning metrics for a market asynchronously.

    Args:
        symbol: Asset ticker symbol (e.g., 'GC', 'ES', 'CL', '6E', 'ZB').
        contract_type: Position breakdown type:
            - `'fut'`: Futures only.
            - `'futopt'`: Combined Futures and Options (default).
            - `'opt'`: Options delta equivalent only.
        lookback_weeks: Number of weekly CFTC reports to retrieve (1 to 156, default 156).
        report: CFTC report family ('legacy', 'disaggregated', 'tff', 'supplemental').
        as_dataframe: When `True` (default), returns a Pandas DataFrame indexed
            by `report_date` in ascending order. When `False`, returns a validated
            `TimeseriesResponse` model.

    Returns:
        A DatetimeIndex-sorted Pandas DataFrame if `as_dataframe=True`, or a
        `TimeseriesResponse` Pydantic model if `as_dataframe=False`.

    Raises:
        AuthenticationError: If API key is missing or invalid (HTTP 401).
        PermissionDeniedError: If account tier lacks access (HTTP 403).
        NotFoundError: If symbol is unrecognized or report data is missing (HTTP 404).
        RateLimitExceededError: If daily request quota is exceeded (HTTP 429).
        ServerError: If upstream COT Bias servers return an error (HTTP 5xx).
        CotApiError: Base exception for any unexpected API errors.

    Example:
        ```python
        import asyncio
        from cotbias import AsyncCotClient

        async def main():
            async with AsyncCotClient() as client:
                df = await client.timeseries.get("GC", lookback_weeks=26)
                print(df.tail())

        asyncio.run(main())
        ```
    """
    params: dict[str, str | int] = {
        "contract_type": contract_type,
        "lookback_weeks": lookback_weeks,
        "report": report,
    }
    resp = await self._client.get(f"/timeseries/{symbol.upper()}", params=params)
    handle_error_response(resp)

    parsed = TimeseriesResponse.model_validate(resp.json())

    if not as_dataframe:
        return parsed

    raw_data = [row.model_dump() for row in parsed.data]
    if not raw_data:
        return cast("pd.DataFrame", pd.DataFrame())

    df = pd.DataFrame(raw_data)
    if "report_date" in df.columns:
        df["report_date"] = pd.to_datetime(df["report_date"])
        df = df.set_index("report_date").sort_index(ascending=True)

    return cast("pd.DataFrame", df)

TimeseriesResource

TimeseriesResource(client: Client)

Synchronous resource for querying historical weekly COT positioning timeseries.

Initialize the timeseries resource with an active HTTP client.

Parameters:

Name	Type	Description	Default
client	Client	Active synchronous HTTP client session.	required

Source code in src/cotbias/resources/timeseries.py

def __init__(self, client: httpx.Client) -> None:
    """Initialize the timeseries resource with an active HTTP client.

    Args:
        client: Active synchronous HTTP client session.
    """
    self._client = client

get

get(symbol: str, contract_type: ContractType = ..., lookback_weeks: int = ..., report: ReportType = ..., as_dataframe: Literal[True] = ...) -> pd.DataFrame

get(symbol: str, contract_type: ContractType = ..., lookback_weeks: int = ..., report: ReportType = ..., as_dataframe: Literal[False] = ...) -> TimeseriesResponse

get(symbol: str, contract_type: str = ..., lookback_weeks: int = ..., report: str = ..., as_dataframe: bool = ...) -> pd.DataFrame | TimeseriesResponse

get(symbol: str, contract_type: str = 'futopt', lookback_weeks: int = 156, report: str = 'legacy', as_dataframe: bool = True) -> pd.DataFrame | TimeseriesResponse

Fetch historical weekly positioning metrics for a market.

Parameters:

Name	Type	Description	Default
symbol	str	Asset ticker symbol (e.g., 'GC', 'ES', 'CL', '6E', 'ZB').	required
contract_type	str	Position breakdown type: - 'fut': Futures only. - 'futopt': Combined Futures and Options (default). - 'opt': Options delta equivalent only.	'futopt'
lookback_weeks	int	Number of weekly CFTC reports to retrieve (1 to 156, default 156).	156
report	str	CFTC report family: - 'legacy': Commercial / Non-Commercial classifications (default). - 'disaggregated': Producer, Swap Dealer, Managed Money, Other. - 'tff': Financial Traders (Dealer, Asset Mgr, Leveraged Money). - 'supplemental': Commodity Index Trader (CIT) reports.	'legacy'
as_dataframe	bool	When True (default), returns a Pandas DataFrame indexed by report_date in ascending order. When False, returns a validated :class:~cotbias.models.timeseries.TimeseriesResponse model.	True

Returns:

Type	Description
DataFrame | TimeseriesResponse	A DatetimeIndex-sorted Pandas DataFrame if as_dataframe=True, or a
DataFrame | TimeseriesResponse	TimeseriesResponse Pydantic model if as_dataframe=False.

Raises:

Type	Description
AuthenticationError	If API key is missing or invalid (HTTP 401).
PermissionDeniedError	If account tier lacks access (HTTP 403).
NotFoundError	If symbol is unrecognized or report data is missing (HTTP 404).
RateLimitExceededError	If daily request quota is exceeded (HTTP 429).
ServerError	If upstream COT Bias servers return an error (HTTP 5xx).
CotApiError	Base exception for any unexpected API errors.

Example

from cotbias import CotClient

client = CotClient()
df = client.timeseries.get("ES", lookback_weeks=52, as_dataframe=True)
print(df[["commercial_net", "cot_52w_commercial"]].tail())

Source code in src/cotbias/resources/timeseries.py

def get(
    self,
    symbol: str,
    contract_type: str = "futopt",
    lookback_weeks: int = 156,
    report: str = "legacy",
    as_dataframe: bool = True,
) -> pd.DataFrame | TimeseriesResponse:
    """Fetch historical weekly positioning metrics for a market.

    Args:
        symbol: Asset ticker symbol (e.g., 'GC', 'ES', 'CL', '6E', 'ZB').
        contract_type: Position breakdown type:
            - `'fut'`: Futures only.
            - `'futopt'`: Combined Futures and Options (default).
            - `'opt'`: Options delta equivalent only.
        lookback_weeks: Number of weekly CFTC reports to retrieve (1 to 156, default 156).
        report: CFTC report family:
            - `'legacy'`: Commercial / Non-Commercial classifications (default).
            - `'disaggregated'`: Producer, Swap Dealer, Managed Money, Other.
            - `'tff'`: Financial Traders (Dealer, Asset Mgr, Leveraged Money).
            - `'supplemental'`: Commodity Index Trader (CIT) reports.
        as_dataframe: When `True` (default), returns a Pandas DataFrame indexed
            by `report_date` in ascending order. When `False`, returns a validated
            :class:`~cotbias.models.timeseries.TimeseriesResponse` model.

    Returns:
        A DatetimeIndex-sorted Pandas DataFrame if `as_dataframe=True`, or a
        `TimeseriesResponse` Pydantic model if `as_dataframe=False`.

    Raises:
        AuthenticationError: If API key is missing or invalid (HTTP 401).
        PermissionDeniedError: If account tier lacks access (HTTP 403).
        NotFoundError: If symbol is unrecognized or report data is missing (HTTP 404).
        RateLimitExceededError: If daily request quota is exceeded (HTTP 429).
        ServerError: If upstream COT Bias servers return an error (HTTP 5xx).
        CotApiError: Base exception for any unexpected API errors.

    Example:
        ```python
        from cotbias import CotClient

        client = CotClient()
        df = client.timeseries.get("ES", lookback_weeks=52, as_dataframe=True)
        print(df[["commercial_net", "cot_52w_commercial"]].tail())
        ```
    """
    params: dict[str, str | int] = {
        "contract_type": contract_type,
        "lookback_weeks": lookback_weeks,
        "report": report,
    }
    resp = self._client.get(f"/timeseries/{symbol.upper()}", params=params)
    handle_error_response(resp)

    parsed = TimeseriesResponse.model_validate(resp.json())

    if not as_dataframe:
        return parsed

    raw_data = [row.model_dump() for row in parsed.data]
    if not raw_data:
        return cast("pd.DataFrame", pd.DataFrame())

    df = pd.DataFrame(raw_data)
    if "report_date" in df.columns:
        df["report_date"] = pd.to_datetime(df["report_date"])
        df = df.set_index("report_date").sort_index(ascending=True)

    return cast("pd.DataFrame", df)