hqg-algorithms 1.0.0

Algorithmic trading research utilities for quant teams

hqg-algorithms

Interfaces and helper types for writing HQG trading strategies.

Install

python3 -m pip install --upgrade pip setuptools wheel
pip install hqg-algorithms

Quick start

Subclass Strategy, declare your universe and cadence, and implement on_data:

from hqg_algorithms import (
    Strategy, Cadence, Slice, PortfolioView,
    BarSize, ExecutionTiming, Signal, TargetWeights,
)

class BuyAndRebalance(Strategy):
    universe = ["SPY", "IEF"]
    cadence = Cadence(bar_size=BarSize.DAILY, execution=ExecutionTiming.CLOSE_TO_CLOSE)

    def on_data(self, data: Slice, portfolio: PortfolioView) -> Signal:
        return TargetWeights({"SPY": 0.6, "IEF": 0.4})

Declaration	Purpose	Required?
universe	List of ticker strings the platform loads for this strategy	Yes
cadence	Bar resolution and execution timing (optional - defaults to daily, close-to-close)	No
on_data()	Return a Signal: TargetWeights(...), Hold(), or Liquidate()	Yes

Important constraints

• universe must be a non-empty list literal of ticker strings (e.g. ["SPY", "IEF"]). Variables, function calls, and concatenation are not supported.
• cadence must be a direct Cadence(...) call with BarSize.X and/or ExecutionTiming.Y keyword arguments. If omitted, it defaults to Cadence(bar_size=BarSize.DAILY, execution=ExecutionTiming.CLOSE_TO_CLOSE).

Slice maps each symbol to a Bar dataclass with typed fields (open, high, low, close, volume). You can access prices via helpers like data.close("SPY"), or grab the full bar with data.bar("SPY") for direct attribute access. PortfolioView gives read-only access to current equity, cash, positions, and weights.

Execution timing

ExecutionTiming controls when your strategy receives data and when the resulting trades fill. Pick the option that matches your signal logic:

ExecutionTiming	on_data fires at	Trades fill at
CLOSE_TO_CLOSE	Bar close	Same bar's close (DEFAULT)
CLOSE_TO_NEXT_OPEN	Bar close	Next bar's open

CLOSE_TO_NEXT_OPEN is the most realistic for intradaily strategies since it avoids look-ahead bias - your signal only uses data that was already available before the trade executes. The other two modes assume you can observe a price and trade at that same price.

Signal types

on_data() returns a Signal that tells the backtester what to do:

Signal	Meaning
TargetWeights({"SPY": 0.6, "IEF": 0.4})	Rebalance to these weights. Omitted symbols are sold to zero. Weights summing to less than 1.0 leave the remainder in cash.
Hold()	Keep the current allocation unchanged.
Liquidate()	Sell all positions and move fully to cash.

Validating strategies

Use validate_strategy to check strategy source code without executing it. It parses the code with ast and verifies that universe, cadence, and on_data are declared correctly (in a way that services expect).

from hqg_algorithms.validate import validate_strategy

source = open("my_strategy.py").read()
errors = validate_strategy(source)

if errors:
    for e in errors:
        print(f"❌ {e}")
else:
    print("✅ Strategy is valid")

validate_strategy returns a list of error strings - an empty list means the strategy is valid. It checks:

Check	Rule
Syntax	Source must be parseable Python
Strategy class	At least one class must define a universe attribute
universe	Must be a non-empty list literal containing only strings
cadence	If present, must be a Cadence(...) call with valid BarSize / ExecutionTiming keyword args
on_data	Must be defined as a method on the strategy class

This is useful for things like editor integrations or pre-submission validation in web UIs where you want fast feedback before sending code to a service.

Example - SMA crossover

from hqg_algorithms import (
    Strategy, Cadence, Slice, PortfolioView,
    BarSize, ExecutionTiming, Signal, TargetWeights, Hold,
)
from collections import deque


class SimpleSMA(Strategy):
    """Go risk-on when SPY is above its 21-day mean, otherwise hold bonds."""

    universe = ["SPY", "BND"]
    cadence = Cadence(bar_size=BarSize.DAILY, execution=ExecutionTiming.CLOSE_TO_CLOSE)

    def __init__(self):
        self._window = 21
        self._q: deque[float] = deque(maxlen=self._window)

    def on_data(self, data: Slice, portfolio: PortfolioView) -> Signal:
        spy_close = data.close("SPY")
        if spy_close is None:
            return Hold()

        self._q.append(spy_close)

        if len(self._q) < self._window:
            return TargetWeights({"BND": 1.0})  # hold bonds while warming up

        sma = sum(self._q) / len(self._q)

        if spy_close > sma:
            return TargetWeights({"SPY": 0.5, "BND": 0.5})  # uptrend
        return TargetWeights({"BND": 1.0})                   # downtrend

Additional docs

• Publishing workflow and release checklist: docs/publishing.md