Plugin

Pair correlation and beta

Rolling correlation, beta and relative strength of the instrument's daily returns against another market, a yield or any feed you have.

IndicatorNo networkDesktop onlyv1.0.0

Get early access Plugin API documentation Built in indicators

About

Markets move together for a while and then stop: gold with real yields, the yen with US rates, an index with its sector. This indicator measures that link while a backtest runs. It pairs the instrument's daily close with a date keyed feed, the same kind ctx.data reads (written by a data source plugin such as the FRED rates curve, or by you), and gives the correlation of their daily changes over a rolling window, the instrument's beta to the feed, and its relative strength over the window.

It is careful about time. Days are UTC days, as feeds are keyed, and only finished days count. A feed stores each value under the first day it was known, usually a day or two after the day it describes. By default (align "observed") the indicator reads the feed's own delay rule from its _meta block and pairs each day of the instrument with the feed's value for that same day, but only from the bar where that value was known; so a pair can arrive a day or two late, and it never arrives early. With align "known" each day is paired with the value known that day instead.

When the feed does not exist, or with align observed its delay rule is not one the kit knows, ctx.plugin_indicator returns None.

Parameters

Name	Type	Default	Range or options	Description
`feed`	text	`""`		Required: the feed's name, as ctx.data reads it.
`field`	text	`""`		For a feed that holds objects: the key of the number to use. Empty for a feed of numbers.
`changes`	choice	`"returns"`	returns, differences	How the feed's daily change is measured: returns for prices, differences for yields, rates and indices like the VIX.
`window`	int	`60`	min 3, max 1000	Day pairs in the rolling window.
`align`	choice	`"observed"`	observed, known	observed pairs each day with the feed's value for that day, from the bar it was known; known pairs it with the value known on that day.

Use it in a strategy

pythonpc = ctx.plugin_indicator("pair_correlation_beta", "correlation",
                          {"feed": "us_dgs10", "changes": "differences", "window": 60})
bars = ctx.bars(1)
# Reading the feed with ctx.data also makes the run fetch it before it starts.
if bars and ctx.data("us_dgs10", bars[-1].ts) is not None:
    if pc is not None and pc.value is not None and abs(pc.value) > 0.6:
        ctx.signal("EXIT")

The feed must exist before the run: install and switch on the plugin that writes it, and read it once with ctx.data in the strategy (or tick "Fetch external data (news and plugin feeds)") so it is fetched first.

What it provides

`value, correlation`Value	Pearson correlation of the instrument's daily returns and the feed's daily changes over the window, -1 to 1; None until the window is full.
`beta`Value	cov(instrument, feed) / var(feed) over the window: the instrument's return per unit of feed change.
`relative_strength`Value	(instrument now / at the window start) / (feed now / at the window start) - 1; only with changes returns, else None.
`overlap_days`Value	How many day pairs the window holds so far, up to window.

Limits

Daily changes only, on UTC days: an intraday strategy sees the figures of finished days. A weekly feed (like COT data) gives one change a week, not a daily series.
Feeds carry their last value over days without a release. On such a day the feed's change is zero, and holidays the writing plugin skipped for a business day rule are not known here, so around them a pair can take the neighbouring value.
Correlation is not causation and changes over time; a correlation from 60 days is a rough estimate.
A plugin indicator is not drawn on the chart and is not available in Blocks, Pine, Rust or C#; it runs on the run's own instrument only.

Preview

json{
  "beta": -0.031,
  "correlation": -0.42,
  "overlap_days": 60,
  "relative_strength": null,
  "value": -0.42
}

A made-up sample that shows the shape of the output. Not vendor data and not a result.

The code

main.py, the file the app installs.

python"""Pair correlation and beta: the run's instrument against any date keyed feed.

A feed is what `ctx.data(name, date)` reads: one value per UTC day, written by
a data source plugin (another market's daily close, a yield, an index) or by
you. This indicator pairs the instrument's daily close with the feed, day by
day, and reports over the last `window` pairs of daily changes:

    value, correlation   Pearson correlation of the two daily changes
    beta                 cov(instrument, feed) / var(feed): how much the
                         instrument moved per unit of feed change
    relative_strength    (instrument now / then) / (feed now / then) - 1 over
                         the window (only with changes="returns")
    overlap_days         how many day pairs the window holds so far

All of it is None until the window is full.

Days are UTC days, as feeds are keyed. Only finished days count: the day the
bar is in has no close yet. The instrument's change is always its return; the
feed's is its return (changes="returns", for prices) or its difference
(changes="differences", for yields, rates and indices like the VIX).

Which feed value goes with which day? A feed stores each value under the first
day it was known, not the day it describes (a close is usually known the next
day). With align="observed" (the default) the indicator reads the feed's own
delay rule from its `_meta` block and pairs the instrument's day D with the
value describing D, which sits under a later day; it waits until the bar has
reached that day, so it never reads a value before it was known. With
align="known" it pairs day D with the value that was known on D.
"""

import math
from collections import deque

from orionquant_engine.plugins.kit import dates, feeds

CHANGES = ("returns", "differences")
ALIGNS = ("observed", "known")


def _field(bar, name):
    return bar[name] if isinstance(bar, dict) else getattr(bar, name)


def _number(value, field):
    """The feed value as a finite float, or None (missing, not a number)."""
    if field:
        value = value.get(field) if isinstance(value, dict) else None
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return None
    value = float(value)
    return value if math.isfinite(value) else None


class PairCorrelationBeta:
    """update(bar), then read the properties listed in the module docstring."""

    def __init__(self, feed, field="", changes="returns", window=60, align="observed"):
        if not feed:
            raise ValueError("feed is required: the name of a feed, as ctx.data reads it")
        if changes not in CHANGES:
            raise ValueError(f"changes must be one of {', '.join(CHANGES)}")
        if align not in ALIGNS:
            raise ValueError(f"align must be one of {', '.join(ALIGNS)}")
        self.window = int(window)
        if self.window < 3:
            raise ValueError("window must be at least 3 days")
        self.field = field
        self.changes = changes
        # Read once: the feed was fetched before the run. Which day of it may be
        # used at a bar is decided in update(), never here.
        self._feed = feeds.read_feed(feed)
        if not self._feed:
            raise ValueError(f"there is no feed {feed!r}: fetch it with its plugin before the run")
        self._rule = None
        if align == "observed":
            meta = feeds.read_meta(feed) or {}
            self._rule = meta.get("delay_rule")
            try:
                feeds.known_from("2024-01-01", self._rule)
            except (TypeError, ValueError):
                raise ValueError(
                    f"feed {feed!r} has no delay rule the kit knows ({self._rule!r}); use align=\"known\"") from None

        self._day = None
        self._close = None
        self._waiting = deque()  # finished days whose feed value is not known yet: (day, close)
        self._last_pair = None  # (close, feed value) of the last day used
        self._pairs = deque(maxlen=self.window)  # (instrument change, feed change)
        self._levels = deque(maxlen=self.window + 1)  # (close, feed value)
        self.correlation = self.beta = self.relative_strength = None

    def _feed_day(self, day):
        """The feed key holding the value that goes with the instrument's `day`."""
        if self._rule is None:
            return day.isoformat()
        return feeds.known_from(day, self._rule)

    def update(self, bar):
        day = dates.as_day(_field(bar, "ts"))
        if day != self._day:
            if self._day is not None:
                self._waiting.append((self._day, self._close))
            self._day = day
        self._close = float(_field(bar, "close"))

        today = day.isoformat()
        changed = False
        while self._waiting:
            done_day, close = self._waiting[0]
            key = self._feed_day(done_day)
            if key > today:
                break  # that value is not known yet at this bar
            self._waiting.popleft()
            value = _number(self._feed.get(key), self.field)
            if value is not None and close > 0:
                changed = self._add(close, value) or changed
        if changed:
            self._compute()

    def _add(self, close, value):
        """One more day with both a close and a feed value. True when a pair was added."""
        last, self._last_pair = self._last_pair, (close, value)
        if last is None:
            self._levels.append((close, value))
            return False
        prev_close, prev_value = last
        if self.changes == "returns":
            if prev_value <= 0 or value <= 0:
                # A price must be positive; start again from this day.
                self._pairs.clear()
                self._levels.clear()
                self._levels.append((close, value))
                return True
            feed_change = value / prev_value - 1.0
        else:
            feed_change = value - prev_value
        self._pairs.append((close / prev_close - 1.0, feed_change))
        self._levels.append((close, value))
        return True

    def _compute(self):
        self.correlation = self.beta = self.relative_strength = None
        n = len(self._pairs)
        if n < self.window:
            return
        xs = [p[0] for p in self._pairs]
        ys = [p[1] for p in self._pairs]
        mx, my = sum(xs) / n, sum(ys) / n
        sxy = sum((x - mx) * (y - my) for x, y in self._pairs)
        sxx = sum((x - mx) ** 2 for x in xs)
        syy = sum((y - my) ** 2 for y in ys)
        if syy > 0:
            self.beta = sxy / syy
        if sxx > 0 and syy > 0:
            self.correlation = max(-1.0, min(1.0, sxy / math.sqrt(sxx * syy)))
        if self.changes == "returns":
            (c0, v0), (c1, v1) = self._levels[0], self._levels[-1]
            self.relative_strength = (c1 / c0) / (v1 / v0) - 1.0

    @property
    def value(self):
        return self.correlation

    @property
    def overlap_days(self):
        return len(self._pairs)


def make(params):
    params = params or {}
    return PairCorrelationBeta(
        feed=params.get("feed", ""),
        field=params.get("field", ""),
        changes=params.get("changes", "returns"),
        window=params.get("window", 60),
        align=params.get("align", "observed"),
    )


def register(api):
    api.register_indicator(
        "correlation",
        "Pair correlation and beta",
        make,
        [
            {"key": "feed", "label": "Feed name", "type": "text", "default": ""},
            {"key": "field", "label": "Field (for feeds of objects)", "type": "text", "default": ""},
            {"key": "changes", "label": "Feed changes", "type": "choice", "default": "returns",
             "options": list(CHANGES)},
            {"key": "window", "label": "Window (days)", "type": "int", "default": 60, "min": 3, "max": 1000},
            {"key": "align", "label": "Align by", "type": "choice", "default": "observed", "options": list(ALIGNS)},
        ],
    )