Indicators¶

The tradetropy.ta package: built-in technical studies, order-flow indicators and the base classes for writing your own.

ta ¶

Indicator ¶

Base class for indicators. Only requires calculate.

calculate(source) source : 1D array [N] (single-source) or 2D [N x K] (multi-source) returns : array [N] (single-band) or [K x N] (multi-band, n_outputs > 1) NaN where the window is not complete

CLASS ATTRIBUTES (override in subclasses)¶

name : str Short lowercase identifier. E.g.: "sma", "rsi", "bb".

IndicatorCategory

Functional classification. Determines the default overlay when plot_config.overlay is None (see _CATEGORY_OVERLAY_DEFAULTS).

list[str]

Names of PRICE output series accessible by the user. Can be class attribute (static) or instance attribute (in init for indicators where names depend on parameters). Do not include timestamp series (those go in ts_band_indices). Examples: ["upper","mid","lower"] for BB, ["pivot_high","pivot_low"] for PivotHighLow.

list[int]

Indices of internal timestamp auxiliary bands. Plotting excludes them from rendering and uses them to reposition markers at the real bar.

list[str]

Names for ts bands, in the same order as ts_band_indices. Enables attribute access from on_data(): self.cpivot.ts_high[-1] -> real timestamp of last PH self.cpivot.ts_low[-1] -> real timestamp of last PL Opt-in: leave [] if explicit names are not needed. Index access (self.cpivot[2][-1]) always works.

IndicatorPlotConfig

Visualization configuration for the indicator. Indicators should assign self.plot_config in init to declare their defaults. add_indicator() can override any field.

METHODS¶

col_name(symbol, col_source) -> str Internal DataStore key. Do not show to user.

display_name() -> str Human-readable name for legends. Used when plot_config.name is None.

plot_config `property` `writable` ¶

plot_config: IndicatorPlotConfig

Visualization config. Subclasses should override this by assigning self.plot_config = IndicatorPlotConfig(...) in init. Default here is an empty config - overlay is resolved by category.

calculate ¶

calculate(source: ndarray) -> np.ndarray

Calculate the indicator over the full array. Must be vectorized.

Source code in src/tradetropy/ta/base.py

def calculate(self, source: np.ndarray) -> np.ndarray:
    """Calculate the indicator over the full array. Must be vectorized."""
    raise NotImplementedError(f"{type(self).__name__}.calculate not implemented")

draw ¶

draw(cfg: 'IndicatorPlotConfig | None' = None, *, interval_ms: 'int | None' = None) -> list

Optional: emit declarative draw primitives for geometry that is not a per-bar series (e.g. a volume-by-price histogram, price-time zones).

Return a list of primitives from :mod:tradetropy.ta.draw (HBars, HLines, Segments, Points, Rects, Labels); a single generic renderer turns them into glyphs for both the static and live charts, so no plotting-package code is needed. The default returns no primitives: ordinary series indicators draw purely from their IndicatorPlotConfig.

Parameters:

Name	Type	Description	Default
`cfg`	`IndicatorPlotConfig \| None`	The effective plot config for this use (after add_indicator() overrides). May be None.	`None`
`interval_ms`	`int \| None`	OHLC interval, available for geometry that needs candle-width anchoring (e.g. VPVR).	`None`

Returns:

Name	Type	Description
`list`	`list`	Draw primitives, or [] when the indicator draws as series only.

Source code in src/tradetropy/ta/base.py

def draw(self, cfg: "IndicatorPlotConfig | None" = None,
         *, interval_ms: "int | None" = None) -> list:
    """
    Optional: emit declarative draw primitives for geometry that is not a
    per-bar series (e.g. a volume-by-price histogram, price-time zones).

    Return a list of primitives from :mod:`tradetropy.ta.draw` (``HBars``,
    ``HLines``, ``Segments``, ``Points``, ``Rects``, ``Labels``); a single
    generic renderer turns them into glyphs for both the static and live
    charts, so no plotting-package code is needed. The default returns no
    primitives: ordinary series indicators draw purely from their
    ``IndicatorPlotConfig``.

    Args:
        cfg (IndicatorPlotConfig | None): The effective plot config for this
            use (after add_indicator() overrides). May be None.
        interval_ms (int | None): OHLC interval, available for geometry that
            needs candle-width anchoring (e.g. VPVR).

    Returns:
        list: Draw primitives, or [] when the indicator draws as series only.
    """
    return []

refs `classmethod` ¶

refs(proxy) -> list

Build the ColumnRef list this indicator consumes, in the expected order.

Generic convenience so a multi-source indicator does not force the caller to spell out every column by hand. Reads cls.source_cols and resolves each name against the proxy:

self.stoch = self.add_indicator(
    Stochastic.refs(self.btc), Stochastic(),
)

Equivalent to passing the proxy directly to add_indicator (which calls :meth:default_refs). Passing an explicit ColumnRef list stays supported. Indicators with tick-specific columns (order flow) override this.

Parameters:

Name	Type	Description	Default
`proxy`	`OhlcProxy \| TickProxy`	An already-subscribed source proxy.	required

Returns:

Type	Description
`list`	list[ColumnRef]: One ColumnRef per name in `source_cols`.

Source code in src/tradetropy/ta/base.py

@classmethod
def refs(cls, proxy) -> list:
    """
    Build the ColumnRef list this indicator consumes, in the expected order.

    Generic convenience so a multi-source indicator does not force the
    caller to spell out every column by hand. Reads ``cls.source_cols`` and
    resolves each name against the proxy:

        self.stoch = self.add_indicator(
            Stochastic.refs(self.btc), Stochastic(),
        )

    Equivalent to passing the proxy directly to ``add_indicator`` (which
    calls :meth:`default_refs`). Passing an explicit ColumnRef list stays
    supported. Indicators with tick-specific columns (order flow) override
    this.

    Args:
        proxy (OhlcProxy | TickProxy): An already-subscribed source proxy.

    Returns:
        list[ColumnRef]: One ColumnRef per name in ``source_cols``.
    """
    return [proxy.col_ref(col) for col in cls.source_cols]

default_refs ¶

default_refs(proxy) -> list

Resolve the source columns when the proxy is passed directly to add_indicator (add_indicator(self.btc, Stochastic())).

Instance-level counterpart of :meth:refs; reads self.source_cols so subclasses that make source_cols an instance attribute (dynamic columns) resolve correctly. Order-flow indicators override this to build tick-specific refs.

Parameters:

Name	Type	Description	Default
`proxy`	`OhlcProxy \| TickProxy`	An already-subscribed source proxy.	required

Returns:

Type	Description
`list`	list[ColumnRef]: One ColumnRef per name in `source_cols`.

Source code in src/tradetropy/ta/base.py

def default_refs(self, proxy) -> list:
    """
    Resolve the source columns when the proxy is passed directly to
    ``add_indicator`` (``add_indicator(self.btc, Stochastic())``).

    Instance-level counterpart of :meth:`refs`; reads ``self.source_cols``
    so subclasses that make ``source_cols`` an instance attribute (dynamic
    columns) resolve correctly. Order-flow indicators override this to build
    tick-specific refs.

    Args:
        proxy (OhlcProxy | TickProxy): An already-subscribed source proxy.

    Returns:
        list[ColumnRef]: One ColumnRef per name in ``source_cols``.
    """
    return [proxy.col_ref(col) for col in self.source_cols]

col_name ¶

col_name(symbol: str, col_source: str = '') -> str

Internal DataStore key. Do not use as display label.

Source code in src/tradetropy/ta/base.py

def col_name(self, symbol: str, col_source: str = "") -> str:
    """Internal DataStore key. Do not use as display label."""
    cls_name = self.name or type(self).__name__.lower()
    length = getattr(self, "length", "")
    suffix = str(length) if length != "" else ""
    return f"{cls_name}{suffix}_{symbol}"

display_name ¶

display_name() -> str

Human-readable name for legends and panel titles.

Examples:

RSI(14) -> "RSI(14)" SMA(20) -> "SMA(20)" BollingerBands() -> "BB(20)" PivotHighLow(3) -> "Pivot(3)"

Source code in src/tradetropy/ta/base.py

def display_name(self) -> str:
    """
    Human-readable name for legends and panel titles.

    Examples:
        RSI(14)          -> "RSI(14)"
        SMA(20)          -> "SMA(20)"
        BollingerBands() -> "BB(20)"
        PivotHighLow(3)  -> "Pivot(3)"
    """
    cls_name = self.name or type(self).__name__
    label = cls_name.upper() if len(cls_name) <= 4 else cls_name.capitalize()
    length = getattr(self, "length", None)
    if length is not None:
        return f"{label}({length})"
    param = getattr(self, "n", None) or getattr(self, "swing", None)
    if param is not None:
        return f"{label}({param})"
    return label

IndicatorPlotConfig `dataclass` ¶

IndicatorPlotConfig(plot: bool = True, overlay: bool | None = None, panel_height: int = 100, panel_title: str | None = None, name: str | list[str] | None = None, zorder: int = 0, color: str | list[str] | None = None, line_width: float | list = 1.8, line_dash: str | list = 'solid', line_alpha: float | list = 0.9, renderer: str | list = 'line', front_dim: int | None = None, marker: str | list = 'circle', marker_size: int | list = 8, marker_alpha: float | list = 0.85, marker_fill: bool | list = True, marker_line_width: float | list = 1.0, bar_width_factor: float | list = 0.8, bar_align: str | list = 'center', bar_alpha: float | list = 0.75, bar_color_positive: str | list | None = None, bar_color_negative: str | list | None = None, show_legend: bool | None = None, reference_lines: list[dict] = list(), autoscale: bool = False, exclude_from_autoscale: bool = False, rect_fill_alpha: float = 0.15, rect_line_alpha: float = 0.6, rect_line_width: float = 1.0, label_font_size: str = '9pt', label_x_offset: int = 5, label_y_offset: int = 5, label_text_align: str = 'left', label_text_baseline: str = 'bottom', arrow_size: int = 10, arrow_length: int = 0)

Complete visualization configuration for an indicator.

Visibility fields¶

plot : False -> do not render in any panel. overlay : True -> on top of OHLC / False -> own panel. None -> resolved from indicator category (see _CATEGORY_OVERLAY_DEFAULTS). panel_height : height in px of own panel (only if overlay=False). panel_title : Y-axis title when indicator is in its own panel. None -> uses name (if str) or display_name() of indicator. name : legend label. - str -> single legend entry controlling all dimensions with one click. - list[str] -> one entry per dimension, independent clicks. - None -> uses display_name() of indicator. zorder : render order within panel (higher = on top).

Line fields (when scatter=False) - scalar or list per dimension¶

color : str | list[str] | None - color per dimension or single for all. None -> automatic color cycle. line_width : float | list[float] - line width per dimension. line_dash : LineDash | list[LineDash] - line style per dimension. line_alpha : float | list[float] - opacity per dimension.

Scatter fields (when scatter=True) - scalar or list per dimension¶

scatter : True -> render as points instead of line. marker : MarkerType | list[MarkerType] - marker shape per dimension. marker_size : int | list[int] - marker size per dimension. marker_alpha : float | list[float] - opacity per dimension. marker_fill : bool | list[bool] - fill per dimension. marker_line_width : float | list[float] - border width per dimension.

Reference lines¶

reference_lines : list of dicts with horizontal lines in the panel. Format: {"value": float, "color": str, "dash": str, "label": str, "width": float}

Per-dimension style usage¶

Bollinger Bands with differentiated lines:

self.bb = self.add_indicator(
    self.btc.close_ref, BollingerBands(20, 2.0),
    name="BB(20)",                              # one legend -> one click
    line_dash=["solid", "dashed", "solid"],
    line_width=[1.5, 1.0, 1.5],
    line_alpha=[0.9, 0.6, 0.9],
    color=["#FF6B35", "#888888", "#3B82F6"],
)

Legend per dimension (independent clicks):

self.bb = self.add_indicator(
    self.btc.close_ref, BollingerBands(20),
    name=["BB Upper", "BB Mid", "BB Lower"],
)

Explicit panel title:

self.rsi = self.add_indicator(
    self.btc.close_ref, RSI(14),
    panel_title="RSI",
)

scatter `property` ¶

scatter: bool

Backwards compatibility: True if all dims use renderer="scatter".

resolve_overlay ¶

resolve_overlay(category: IndicatorCategory) -> bool

Returns the effective overlay value, resolving None from the category.

Source code in src/tradetropy/ta/base.py

def resolve_overlay(self, category: IndicatorCategory) -> bool:
    """
    Returns the effective overlay value, resolving None from the category.
    """
    if self.overlay is not None:
        return self.overlay
    return _CATEGORY_OVERLAY_DEFAULTS.get(category, False)

dim ¶

dim(field_name: str, idx: int)

Returns the field value for dimension idx. Delegates to _resolve_dim() - supports scalar and list.

Example

pc.dim("line_dash", 0) -> "solid" pc.dim("line_dash", 1) -> "dashed" (if line_dash is a list)

Source code in src/tradetropy/ta/base.py

def dim(self, field_name: str, idx: int):
    """
    Returns the field value for dimension `idx`.
    Delegates to _resolve_dim() - supports scalar and list.

    Example:
        pc.dim("line_dash", 0)  -> "solid"
        pc.dim("line_dash", 1)  -> "dashed"  (if line_dash is a list)
    """
    return _resolve_dim(getattr(self, field_name), idx)

SMA ¶

SMA(length: int)

Bases: Indicator

Simple Moving Average. source: close [N]

Usage on ticks

self.sma_price = self.add_indicator(self.btc.price_ref, SMA(length=20))

Usage on OHLC candles

self.btc_1m = self.subscribe_ohlc("BTCUSDT", interval_min=1) self.sma_close = self.add_indicator(self.btc_1m.close_ref, SMA(length=10))

Source code in src/tradetropy/ta/trend.py

def __init__(self, length: int):
    self.length = length
    self.plot_config = IndicatorPlotConfig()

EMA ¶

EMA(length: int)

Bases: Indicator

Exponential Moving Average. alpha = 2 / (length + 1). Seed: SMA of the first length bars. source: close [N]

Usage

self.ema_close = self.add_indicator(self.btc_1m.close_ref, EMA(length=20))

Source code in src/tradetropy/ta/trend.py

def __init__(self, length: int):
    self.length = length
    self._alpha = 2.0 / (length + 1)
    self.plot_config = IndicatorPlotConfig()

MACD ¶

MACD(fast: int = 12, slow: int = 26, signal: int = 9)

Bases: Indicator

Moving Average Convergence Divergence. Produces 3 series: macd, signal, histogram. source: close [N]

Parameters:

Name	Description	Default
`fast`	fast EMA period (default 12)	required
`slow`	slow EMA period (default 26)	required
`signal`	signal EMA period over MACD line (default 9)	required

Usage

self.btc = self.subscribe_ohlc("BTCUSDT", timeframe='1m') self.macd = self.add_indicator( self.btc.close_ref, MACD(12, 26, 9), plot=True, overlay=False, )

Access in on_data(): self.macd.macd[-1] -> MACD value of current (partial) candle self.macd.signal[-1] -> signal line self.macd.histogram[-1] -> histogram (MACD - Signal)

# Bullish crossover:
if self.macd.macd[-2] < self.macd.signal[-2] and \
   self.macd.macd[-1] > self.macd.signal[-1]:
    ...  # golden cross

Note

The seed of each EMA is the SMA of the first length bars, same as the EMA in this module.
min_periods = slow + signal - 1 (minimum to have a valid histogram)
The histogram is rendered as colored bars (green/red).

Source code in src/tradetropy/ta/trend.py

def __init__(
    self,
    fast: int = 12,
    slow: int = 26,
    signal: int = 9,
):
    self.fast = fast
    self.slow = slow
    self.signal_period = signal
    self.length = slow

    self.output_names = ["macd", "signal", "histogram"]

    self.plot_config = IndicatorPlotConfig(
        overlay=False,
        panel_height=100,
        panel_title="MACD",
        autoscale=True,
        color=["#2563EB", "#F59E0B", "#888888"],
        line_width=[1.6, 1.2, 1.0],
        renderer=["line", "line", "bar"],
        bar_color_positive="#0ECB81",
        bar_color_negative="#F6465D",
        bar_alpha=0.75,
        reference_lines=[
            {"value": 0.0, "color": "#888888", "dash": "dashed",
             "label": "", "width": 0.8},
        ],
    )

RSI ¶

RSI(length: int = 14)

Bases: Indicator

Relative Strength Index. source: close [N]

Source code in src/tradetropy/ta/momentum.py

def __init__(self, length: int = 14):
    self.length = length
    self.output_names = ["rsi"]
    self.plot_config = IndicatorPlotConfig(
        overlay=False,
        panel_height=100,
        panel_title="RSI",
        reference_lines=[
            {"value": 70.0, "color": "#F6465D", "dash": "dashed",
             "label": "70", "width": 0.8},
            {"value": 30.0, "color": "#0ECB81", "dash": "dashed",
             "label": "30", "width": 0.8},
        ],
    )

calculate ¶

calculate(source: ndarray) -> np.ndarray

Wilder RSI. O(N) with Python loop.

Source code in src/tradetropy/ta/momentum.py

def calculate(self, source: np.ndarray) -> np.ndarray:
    """Wilder RSI. O(N) with Python loop."""
    n = len(source)
    result = np.full(n, np.nan, dtype=np.float64)
    L = self.length
    if n < L + 2:
        return result

    delta  = np.diff(source.astype(np.float64))
    gains  = np.where(delta > 0, delta, 0.0)
    losses = np.where(delta < 0, -delta, 0.0)

    avg_g = float(np.mean(gains[:L]))
    avg_l = float(np.mean(losses[:L]))

    for i in range(L, len(delta)):
        avg_g = (avg_g * (L - 1) + gains[i]) / L
        avg_l = (avg_l * (L - 1) + losses[i]) / L
        rs    = avg_g / avg_l if avg_l > 0 else np.inf
        result[i + 1] = 100.0 - (100.0 / (1.0 + rs))

    return result

ATR ¶

ATR(length: int = 14)

Bases: Indicator

Average True Range. source: HLC [N×3]

Usage

self.bb = self.add_indicator( self.btc.close_ref, BollingerBands(20, 2.0), name=["BB Upper", "BB Mid", "BB Lower"], plot=True, overlay=True, color=["#FF6B35", "#888888", "#3B82F6"], )

Access

self.bb.upper[-1] -> upper band of partial candle self.bb.mid[-1] -> mid (SMA) self.bb.lower[-1] -> lower band self.bb[0][-1] -> equivalent to upper

Source code in src/tradetropy/ta/volatility.py

def __init__(self, length: int = 14):
    self.length = length
    self.plot_config = IndicatorPlotConfig(overlay=False)

BollingerBands ¶

BollingerBands(length: int = 20, std_dev: float = 2.0)

Bases: Indicator

Bollinger Bands. Produces 3 series: upper, mid, lower. source: close [N]

Usage

self.bb = self.add_indicator( self.btc.close_ref, BollingerBands(20, 2.0), name=["BB Upper", "BB Mid", "BB Lower"], plot=True, overlay=True, color=["#FF6B35", "#888888", "#3B82F6"], )

Access

self.bb.upper[-1] -> upper band of partial candle self.bb.mid[-1] -> mid (SMA) self.bb.lower[-1] -> lower band self.bb[0][-1] -> equivalent to upper

Source code in src/tradetropy/ta/volatility.py

def __init__(self, length: int = 20, std_dev: float = 2.0):
    self.length = length
    self.std_dev = std_dev
    self.output_names = ["upper", "mid", "lower"]
    self.plot_config = IndicatorPlotConfig(
        color=["#3B82F6", "#3B82F6", "#3B82F6"],
        line_dash=["solid", "dashed", "solid"],
        line_width=[1.5, 1, 1.5],
    )

VolumeProfile ¶

VolumeProfile(period='1d', tick_size: float | None = None, bins: int = 100, value_area_pct: float = 0.7, va_recompute_every: int = 1, view: str = 'session', nodes: str | None = None, node_prominence: float = 0.1, max_nodes: int | None = None, anchor=None, buy_color: str = DEFAULT_VP_BUY, sell_color: str = DEFAULT_VP_SELL, poc_color: str = DEFAULT_VP_POC, hvn_color: str = DEFAULT_VP_HVN, lvn_color: str = DEFAULT_VP_LVN)

Bases: _VolumeProfileBase

Kline-based volume profile (uniform high-low distribution).

Source columns (use VolumeProfile.refs(ohlc_proxy)): [ts, open, high, low, close, volume]

Each candle's volume is spread uniformly across the price levels between its low and high. Aggressor side is approximated from candle direction: a bullish candle (close >= open) counts as ask (buy) volume, a bearish one as bid (sell) volume.

Source code in src/tradetropy/ta/volume.py

def __init__(
    self,
    period="1d",
    tick_size: float | None = None,
    bins: int = 100,
    value_area_pct: float = 0.70,
    va_recompute_every: int = 1,
    view: str = "session",
    nodes: str | None = None,
    node_prominence: float = 0.1,
    max_nodes: int | None = None,
    anchor=None,
    buy_color: str = DEFAULT_VP_BUY,
    sell_color: str = DEFAULT_VP_SELL,
    poc_color: str = DEFAULT_VP_POC,
    hvn_color: str = DEFAULT_VP_HVN,
    lvn_color: str = DEFAULT_VP_LVN,
):
    """
    Initialize a volume profile indicator.

    Args:
        period (str or int): Reset period for the profile (e.g. "1d", "1w",
            or milliseconds). Each period accumulates its own histogram.
        tick_size (float or None): Price level size. If None it is inferred
            from the price range and ``bins``.
        bins (int): Target number of levels across the range, used only when
            ``tick_size`` is None.
        value_area_pct (float): Fraction of volume defining the value area.
        va_recompute_every (int): Recompute the value area every N rows.
            1 = exact developing VAH/VAL on every row.
        view (str): How the histogram is laid out when plotting.
            "session" (default) draws one profile per period anchored at the
            period start and growing right (TradingView VPSV). "visible"
            aggregates every level across all periods into a single profile
            anchored at the right edge and growing left (TradingView VPVR).
        nodes (str or None): Enable HVN/LVN node detection accessible from
            on_data() via the proxy (``self.vp.hvn`` / ``.lvn`` / ``.nodes``).
            None (default) disables it; accessing the proxy nodes then raises
            ConfigError. 'hvn', 'lvn' or 'both' select which kinds to detect.
        node_prominence (float): Minimum node strength (0..1) to keep.
        max_nodes (int or None): Optional cap per node kind (strongest first).
        anchor: Where each period begins. None or 'utc' (default) keeps the
            legacy Unix-aligned cut (UTC midnight for '1d', Thursday-UTC for
            '1w'; intraday timeframes already fall on round clock marks).
            'data' starts the grid on the first row (e.g. 11:11 -> 11:11 next
            day). A timezone (str like 'America/New_York' or tzinfo) anchors
            to that zone's local midnight, DST-aware. A 2-tuple
            ``(tz, 'HH:MM')`` anchors to a session open time of day (e.g.
            ``('America/New_York', '17:00')`` for CME Globex). Only affects
            period-based cuts; intraday utc behaviour is unchanged.
        buy_color (str): Color of the buy-aggressor (ask) volume segment in
            the histogram. Fixed default, independent of the plot theme.
        sell_color (str): Color of the sell-aggressor (bid) volume segment.
        poc_color (str): Color of the point-of-control line.
        hvn_color (str): Color of the high-volume node markers.
        lvn_color (str): Color of the low-volume node markers.
    """
    if view not in ("session", "visible"):
        raise ValueError(f"view must be 'session' or 'visible', not {view!r}")
    if nodes not in (None, "hvn", "lvn", "both"):
        raise ConfigError(
            f"nodes must be None, 'hvn', 'lvn' or 'both', not {nodes!r}"
        )
    self.period = period
    self.period_ms = parse_timeframe(period)
    self.tick_size = tick_size
    self.bins = bins
    self.value_area_pct = value_area_pct
    self.va_recompute_every = max(int(va_recompute_every), 1)
    self.view = view
    self.nodes = nodes
    self.node_prominence = node_prominence
    self.max_nodes = max_nodes
    self.anchor = anchor
    # Semantic histogram colors, theme-independent (live on the object).
    self.buy_color = buy_color
    self.sell_color = sell_color
    self.poc_color = poc_color
    self.hvn_color = hvn_color
    self.lvn_color = lvn_color
    # Validate the anchor eagerly so configuration errors surface at
    # construction time, not deep inside calculate. The dummy timestamp
    # only exercises the parsing/tz-resolution path.
    try:
        resolve_period_ids(
            np.array([0], dtype=np.int64), self.period_ms, anchor
        )
    except ValueError as exc:
        raise ConfigError(f"invalid anchor: {exc}") from exc

    self.output_names = ["poc", "vah", "val"]

    # Per-period histograms, filled by calculate for the plotting layer.
    self.profiles_: list[dict] = []
    self.tick_size_used_: float | None = None
    # pid -> period-start ts (ms), filled during calculate so that
    # period_profiles() can reconstruct the anchored x extent even when the
    # period grid is not Unix aligned (data/timezone/session anchors).
    self._pid_start_ms: dict[int, int] = {}

    self.plot_config = IndicatorPlotConfig(
        overlay=True,
        renderer="step",
        color=[self.poc_color, "#9CA3AF", "#9CA3AF"],
        line_width=[1.6, 1.0, 1.0],
        line_dash=["solid", "dashed", "dashed"],
        line_alpha=[0.95, 0.7, 0.7],
        front_dim=0,
    )

refs `staticmethod` ¶

refs(ohlc_proxy)

Build the ColumnRef list for this indicator in the expected order.

Parameters:

Name	Type	Description	Default
`ohlc_proxy`	`OhlcProxy`	Proxy returned by subscribe_ohlc().	required

Returns:

Type	Description
	list[ColumnRef]: [ts, open, high, low, close, volume] refs.

Source code in src/tradetropy/ta/volume.py

@staticmethod
def refs(ohlc_proxy):
    """
    Build the ColumnRef list for this indicator in the expected order.

    Args:
        ohlc_proxy (OhlcProxy): Proxy returned by subscribe_ohlc().

    Returns:
        list[ColumnRef]: [ts, open, high, low, close, volume] refs.
    """
    return [
        ohlc_proxy.ts_ref,
        ohlc_proxy.open_ref,
        ohlc_proxy.high_ref,
        ohlc_proxy.low_ref,
        ohlc_proxy.close_ref,
        ohlc_proxy.volume_ref,
    ]

RollingVolumeProfile ¶

RollingVolumeProfile(length: int = 200, tick_size: float | None = None, bins: int = 100, value_area_pct: float = 0.7, va_recompute_every: int = 1, nodes: str | None = None, node_prominence: float = 0.1, max_nodes: int | None = None, buy_color: str = DEFAULT_VP_BUY, sell_color: str = DEFAULT_VP_SELL, poc_color: str = DEFAULT_VP_POC, hvn_color: str = DEFAULT_VP_HVN, lvn_color: str = DEFAULT_VP_LVN)

Bases: VolumeProfile

Kline-based volume profile over a sliding window of the last length bars. source: ts, open, high, low, close, volume [N×6]

Unlike VolumeProfile (which resets every period and gives a saw-tooth developing series), this profile never resets: at each bar it aggregates the trailing length candles, producing continuous POC / VAH / VAL lines suitable for plotting and for reading in on_data()::

self.rvp = self.add_indicator(self.btc, RollingVolumeProfile(length=200))
# in on_data():
if self.btc.close[-1] > self.rvp.vah[-1]:
    ...

The single histogram exposed for plotting is the profile of the current (most recent) window, drawn at the right edge like TradingView's VPVR.

Initialize a rolling volume profile.

Parameters:

Name	Type	Description	Default
`length`	`int`	Number of trailing candles aggregated into the profile.	`200`
`tick_size`	`float or None`	Price level size; inferred when None.	`None`
`bins`	`int`	Target number of levels, used only when tick_size is None.	`100`
`value_area_pct`	`float`	Fraction of volume defining the value area.	`0.7`
`va_recompute_every`	`int`	Recompute the value area every N bars.	`1`
`nodes`	`str or None`	HVN/LVN detection (see _VolumeProfileBase). The nodes cover the trailing `length` bars ending at the current bar (the same window as the developing profile).	`None`
`node_prominence`	`float`	Minimum node strength (0..1) to keep.	`0.1`
`max_nodes`	`int or None`	Optional cap per node kind.	`None`
`buy_color`	`str`	Color of the buy-aggressor (ask) volume segment.	`DEFAULT_VP_BUY`
`sell_color`	`str`	Color of the sell-aggressor (bid) volume segment.	`DEFAULT_VP_SELL`
`poc_color`	`str`	Color of the point-of-control line.	`DEFAULT_VP_POC`
`hvn_color`	`str`	Color of the high-volume node markers.	`DEFAULT_VP_HVN`
`lvn_color`	`str`	Color of the low-volume node markers.	`DEFAULT_VP_LVN`

Source code in src/tradetropy/ta/volume.py

def __init__(
    self,
    length: int = 200,
    tick_size: float | None = None,
    bins: int = 100,
    value_area_pct: float = 0.70,
    va_recompute_every: int = 1,
    nodes: str | None = None,
    node_prominence: float = 0.1,
    max_nodes: int | None = None,
    buy_color: str = DEFAULT_VP_BUY,
    sell_color: str = DEFAULT_VP_SELL,
    poc_color: str = DEFAULT_VP_POC,
    hvn_color: str = DEFAULT_VP_HVN,
    lvn_color: str = DEFAULT_VP_LVN,
):
    """
    Initialize a rolling volume profile.

    Args:
        length (int): Number of trailing candles aggregated into the profile.
        tick_size (float or None): Price level size; inferred when None.
        bins (int): Target number of levels, used only when tick_size is None.
        value_area_pct (float): Fraction of volume defining the value area.
        va_recompute_every (int): Recompute the value area every N bars.
        nodes (str or None): HVN/LVN detection (see _VolumeProfileBase). The
            nodes cover the trailing ``length`` bars ending at the current
            bar (the same window as the developing profile).
        node_prominence (float): Minimum node strength (0..1) to keep.
        max_nodes (int or None): Optional cap per node kind.
        buy_color (str): Color of the buy-aggressor (ask) volume segment.
        sell_color (str): Color of the sell-aggressor (bid) volume segment.
        poc_color (str): Color of the point-of-control line.
        hvn_color (str): Color of the high-volume node markers.
        lvn_color (str): Color of the low-volume node markers.
    """
    super().__init__(
        period="1d",
        tick_size=tick_size,
        bins=bins,
        value_area_pct=value_area_pct,
        va_recompute_every=va_recompute_every,
        view="visible",
        nodes=nodes,
        node_prominence=node_prominence,
        max_nodes=max_nodes,
        buy_color=buy_color,
        sell_color=sell_color,
        poc_color=poc_color,
        hvn_color=hvn_color,
        lvn_color=lvn_color,
    )
    self.length = max(int(length), 1)
    self._final_profile: dict | None = None
    self._win_ts_start = 0
    self._win_ts_end = 0

period_profiles ¶

period_profiles() -> list[dict]

Single render-ready profile of the most recent window (VPVR-style).

Returns a one-element list (or empty) shaped like the per-period dicts consumed by the plotting layer, so build_volume_profile_source() can draw it at the right edge via view="visible".

Source code in src/tradetropy/ta/volume.py

def period_profiles(self) -> list[dict]:
    """
    Single render-ready profile of the most recent window (VPVR-style).

    Returns a one-element list (or empty) shaped like the per-period dicts
    consumed by the plotting layer, so build_volume_profile_source() can draw
    it at the right edge via view="visible".
    """
    from tradetropy.models.footprint import _FP_LEVEL_COL, _FP_SCALAR_COL

    final = self._final_profile
    if not final:
        return []
    levels = final["levels"]
    if levels.size == 0 or len(levels) == 0:
        return []
    scal = final["scalars"]
    volumes = levels[:, _FP_LEVEL_COL["vol_total"]]
    return [{
        "ts_start": self._win_ts_start,
        "ts_end": self._win_ts_end,
        "poc": float(scal[_FP_SCALAR_COL["poc_price"]]),
        "vah": float(scal[_FP_SCALAR_COL["vah"]]),
        "val": float(scal[_FP_SCALAR_COL["val"]]),
        "max_vol": float(volumes.max()) if len(volumes) else 0.0,
        "prices": levels[:, _FP_LEVEL_COL["price"]],
        "volumes": volumes,
        "vol_bid": levels[:, _FP_LEVEL_COL["vol_bid"]],
        "vol_ask": levels[:, _FP_LEVEL_COL["vol_ask"]],
        "deltas": levels[:, _FP_LEVEL_COL["delta"]],
    }]

LargeTrades ¶

LargeTrades(threshold='p99', by: str = 'volume', window: int = 2000, aggregate_ms: int = 0, min_gap_ms: int = 0, scale: str = 'sqrt', min_size: float = 10.0, max_size: float = 40.0, buy_color: str = DEFAULT_BUY_COLOR, sell_color: str = DEFAULT_SELL_COLOR, fill: bool = False, fill_alpha: float | None = None, label_color: str | None = None, line_width: float = 1.6, label: 'str | None' = 'volume', label_font_size: str = '7pt')

Bases: Indicator

Large-trade detector and bubble overlay (order flow).

Source columns (use LargeTrades.refs(tick_proxy) or pass the tick proxy directly): [ts, price, volume, flags, bid, ask].

Outputs - [4 x N] (1 price band + 3 auxiliary bands):

Price band (accessible by the user, drawn as bubbles): row 0 : price - price of a detected large trade at this tick (NaN otherwise).

Auxiliary bands (excluded from rendering, exposed for on_data() access via ts_output_names): row 1 : volume - traded volume of the large trade (NaN otherwise). row 2 : notional - price * volume of the large trade (NaN otherwise). row 3 : side - aggressor side (+1 buy / -1 sell), NaN otherwise.

Parameters:

Name	Type	Description	Default
`threshold`	`float \| str`	What makes a trade "large". - float: absolute magnitude threshold. - 'pXX': trailing quantile (e.g. 'p99' = top 1% within `window`). - 'Nx': N times the trailing median (e.g. '5x').	`'p99'`
`by`	`str`	Magnitude metric: 'volume', 'notional' (price * volume) or 'delta' (net aggression). 'delta' is only meaningful with `aggregate_ms` > 0, where buys and sells inside a burst net out.	`'volume'`
`window`	`int`	Trailing window (in events) for relative thresholds.	`2000`
`aggregate_ms`	`int`	Execution-burst window in ms. When > 0, consecutive trades are merged before thresholding (same-side for volume/notional, net signed for delta); 0 disables aggregation (per-trade detection).	`0`
`min_gap_ms`	`int`	Minimum spacing in ms between detections (anti-cluster; 0 disables it).	`0`
`scale`	`str`	Bubble sizing: 'sqrt' (area proportional, default) or 'linear'.	`'sqrt'`
`min_size`	`float`	Smallest bubble diameter in px.	`10.0`
`max_size`	`float`	Largest bubble diameter in px.	`40.0`
`buy_color`	`str`	Color for buy-aggressor bubbles (default blue).	`DEFAULT_BUY_COLOR`
`sell_color`	`str`	Color for sell-aggressor bubbles (default pink).	`DEFAULT_SELL_COLOR`
`fill`	`bool`	True fills the circle, False border only (hollow).	`False`
`fill_alpha`	`float`	Fill transparency (0.0-1.0). Only applies when fill=True. None = uses default alpha (0.35).	`None`
`label_color`	`str \| None`	Color of the internal labels (magnitude labels). None = uses the same color as the circle border.	`None`
`line_width`	`float`	Border width of the hollow circles.	`1.6`
`label`	`str \| None`	Magnitude number drawn next to each bubble: 'volume', 'notional', 'delta', or None to hide labels. The label always shows the trade/burst's real magnitude (not a sequential index).	`'volume'`
`label_font_size`	`str`	Font size for the magnitude labels.	`'7pt'`

Source code in src/tradetropy/ta/order_flow/large_trades.py

def __init__(
    self,
    threshold="p99",
    by: str = "volume",
    window: int = 2000,
    aggregate_ms: int = 0,
    min_gap_ms: int = 0,
    scale: str = "sqrt",
    min_size: float = 10.0,
    max_size: float = 40.0,
    buy_color: str = DEFAULT_BUY_COLOR,
    sell_color: str = DEFAULT_SELL_COLOR,
    fill: bool = False,
    fill_alpha: float | None = None,
    label_color: str | None = None,
    line_width: float = 1.6,
    label: "str | None" = "volume",
    label_font_size: str = "7pt",
):
    if by not in ("volume", "notional", "delta"):
        raise ValueError(
            f"by must be 'volume', 'notional' or 'delta', not {by!r}"
        )
    if label not in (None, "volume", "notional", "delta"):
        raise ValueError(
            f"label must be None, 'volume', 'notional' or 'delta', "
            f"not {label!r}"
        )
    if scale not in ("sqrt", "linear"):
        raise ValueError(f"scale must be 'sqrt' or 'linear', not {scale!r}")

    self.threshold = threshold
    self.by = by
    self.window = int(window)
    self.aggregate_ms = int(aggregate_ms)
    self.min_gap_ms = int(min_gap_ms)
    self.scale = scale
    self.min_size = float(min_size)
    self.max_size = float(max_size)
    self.buy_color = buy_color
    self.sell_color = sell_color
    self.fill = bool(fill)
    self.fill_alpha = float(fill_alpha) if fill_alpha is not None else None
    self.label_color = label_color
    self.line_width = float(line_width)
    self.label = label
    self.label_font_size = label_font_size

    # length drives the engine warmup; relative thresholds need the window.
    self.length = self.window if isinstance(threshold, str) else 1

    # Filled by calculate for the plotting layer.
    self._deep_events: dict = {}
    self._deep_style: dict = self._build_style()

    self.plot_config = IndicatorPlotConfig(
        overlay=True,
        exclude_from_autoscale=True,
        renderer="none",   # draw-only: bubbles come from draw(), no series line
        name="Large Trades",
    )

refs `staticmethod` ¶

refs(tick_proxy)

Build the ColumnRef list for this indicator in the expected order.

Parameters:

Name	Type	Description	Default
`tick_proxy`	`TickProxy`	Proxy returned by subscribe_ticks().	required

Returns:

Type	Description
	list[ColumnRef]: [ts, price, volume, flags, bid, ask] refs.

Source code in src/tradetropy/ta/order_flow/large_trades.py

@staticmethod
def refs(tick_proxy):
    """
    Build the ColumnRef list for this indicator in the expected order.

    Args:
        tick_proxy (TickProxy): Proxy returned by subscribe_ticks().

    Returns:
        list[ColumnRef]: [ts, price, volume, flags, bid, ask] refs.
    """
    return [
        tick_proxy.col_ref("ts"),
        tick_proxy.price_ref,
        tick_proxy.col_ref("volume"),
        tick_proxy.col_ref("flags"),
        tick_proxy.bid_ref,
        tick_proxy.ask_ref,
    ]

default_refs ¶

default_refs(proxy)

ColumnRefs resolved from a proxy so add_indicator(proxy, ind) works.

Source code in src/tradetropy/ta/order_flow/large_trades.py

def default_refs(self, proxy):
    """ColumnRefs resolved from a proxy so add_indicator(proxy, ind) works."""
    return type(self).refs(proxy)

detect ¶

detect(source: ndarray) -> dict

Run causal detection over a [N x 6] source matrix.

Shared by calculate (backtest) and the live streaming path so both flag the exact same trades.

Parameters:

Name	Type	Description	Default
`source`	`ndarray`	[N x 6] columns [ts, price, volume, flags, bid, ask].	required

Returns:

Name	Type	Description
`dict`	`dict`	`detect_large_trades` result with keys 'mask', 'metric',
	`dict`	'side' and 'events'.

Source code in src/tradetropy/ta/order_flow/large_trades.py

def detect(self, source: np.ndarray) -> dict:
    """
    Run causal detection over a [N x 6] source matrix.

    Shared by calculate (backtest) and the live streaming path so both
    flag the exact same trades.

    Args:
        source (np.ndarray): [N x 6] columns [ts, price, volume, flags,
            bid, ask].

    Returns:
        dict: ``detect_large_trades`` result with keys 'mask', 'metric',
        'side' and 'events'.
    """
    ts = source[:, 0].astype(np.float64)
    price = source[:, 1].astype(np.float64)
    volume = source[:, 2].astype(np.float64)
    flags = source[:, 3] if source.shape[1] > 3 else None
    bid = source[:, 4] if source.shape[1] > 4 else None
    ask = source[:, 5] if source.shape[1] > 5 else None
    return detect_large_trades(
        ts, price, volume, flags=flags, bid=bid, ask=ask,
        threshold=self.threshold, by=self.by, window=self.window,
        min_gap_ms=self.min_gap_ms, aggregate_ms=self.aggregate_ms,
    )

calculate ¶

calculate(source: ndarray) -> np.ndarray

Detect large trades and emit the [4 x N] band matrix.

Bands are anchored at each detected event's representative tick (the last trade of an aggregated burst) and carry the burst's aggregated magnitudes, so self.volume / self.notional reflect the merged execution, not a single child print.

Parameters:

Name	Type	Description	Default
`source`	`ndarray`	[N x 6] columns [ts, price, volume, flags, bid, ask].	required

Returns:

Type	Description
`ndarray`	np.ndarray: [4 x N] - row 0 price (NaN unless large), rows 1-3
`ndarray`	volume / notional / side (NaN unless large).

Source code in src/tradetropy/ta/order_flow/large_trades.py

def calculate(self, source: np.ndarray) -> np.ndarray:
    """
    Detect large trades and emit the [4 x N] band matrix.

    Bands are anchored at each detected event's representative tick (the
    last trade of an aggregated burst) and carry the burst's aggregated
    magnitudes, so ``self.volume`` / ``self.notional`` reflect the merged
    execution, not a single child print.

    Args:
        source (np.ndarray): [N x 6] columns [ts, price, volume, flags,
            bid, ask].

    Returns:
        np.ndarray: [4 x N] - row 0 price (NaN unless large), rows 1-3
        volume / notional / side (NaN unless large).
    """
    if source.ndim != 2 or source.shape[1] < 3 or len(source) == 0:
        self._deep_events = {}
        return np.full((4, 0), np.nan, dtype=np.float64)

    n = len(source)
    res = self.detect(source)
    mask = res["mask"]
    events = res["events"]

    out = np.full((4, n), np.nan, dtype=np.float64)
    idx = np.where(mask)[0]
    if idx.size:
        out[0, idx] = events["price"]
        out[1, idx] = events["volume"]
        out[2, idx] = events["notional"]
        out[3, idx] = events["side"].astype(np.float64)

    self._deep_events = {
        "ts": np.asarray(events["ts"], dtype=np.int64),
        "price": np.asarray(events["price"], dtype=np.float64),
        "volume": np.asarray(events["volume"], dtype=np.float64),
        "notional": np.asarray(events["notional"], dtype=np.float64),
        "delta": np.asarray(events["delta"], dtype=np.float64),
        "side": np.asarray(events["side"], dtype=np.int8),
        "metric": np.asarray(events["metric"], dtype=np.float64),
    }
    self._deep_style = self._build_style()
    return out

event_labels ¶

event_labels() -> list

Build the magnitude label strings for the detected events.

Returns:

Type	Description
`list`	list[str]: One compact magnitude string per detected trade, or an
`list`	empty list when `label` is None.

Source code in src/tradetropy/ta/order_flow/large_trades.py

def event_labels(self) -> list:
    """
    Build the magnitude label strings for the detected events.

    Returns:
        list[str]: One compact magnitude string per detected trade, or an
        empty list when ``label`` is None.
    """
    if not self.label or not self._deep_events:
        return []
    src = self._deep_events.get(self.label)
    if src is None:
        return []
    return [format_magnitude(float(v)) for v in src]

draw ¶

draw(cfg=None, *, interval_ms=None) -> dict

Emit the large-trade bubbles as a Points primitive plus magnitude Labels.

Bubbles are hollow circles (border only) sized per trade and colored by aggressor side; one label per bubble shows the real magnitude. Colors come from the indicator object (buy_color / sell_color), theme-independent.

Source code in src/tradetropy/ta/order_flow/large_trades.py

def draw(self, cfg=None, *, interval_ms=None) -> dict:
    """
    Emit the large-trade bubbles as a Points primitive plus magnitude Labels.

    Bubbles are hollow circles (border only) sized per trade and colored by
    aggressor side; one label per bubble shows the real magnitude. Colors
    come from the indicator object (buy_color / sell_color), theme-independent.
    """
    from tradetropy.ta.draw import Points, Labels
    from tradetropy.ta.order_flow._core import build_bubble_columns

    events = getattr(self, "_deep_events", {}) or {}
    style = getattr(self, "_deep_style", None) or self._build_style()
    cols = build_bubble_columns(events, style, self.event_labels())
    if len(cols["ts"]) == 0:
        return {}

    prims: dict[str, list] = {}
    prims["Large Trades"] = [Points(
        x=list(cols["ts"]), y=list(cols["price"]),
        color=list(cols["color"]), alpha=0.95,
        size=list(cols["size"]), marker="circle",
        fill=style.get("fill", False),
        fill_alpha=style.get("fill_alpha"),
        line_width=style.get("line_width", 1.6),
    )]
    if style.get("label"):
        lbl_color = style.get("label_color") or list(cols["color"])
        prims["Large Trades Labels"] = [Labels(
            x=list(cols["ts"]), y=list(cols["price"]),
            text=list(cols["text"]), color=lbl_color,
            font_size=style.get("label_font_size", "8pt"),
            x_offset=0, y_offset=0,
            text_align="center", text_baseline="middle",
        )]
    return prims

live_refresh ¶

live_refresh(proxy) -> None

Re-run causal detection over the source tick window (live mode).

Tick-mounted indicators are not recomputed by the standard live indicator path (which reads OHLC ring bands), so the generic live primitive updater calls this hook before draw(): it rebuilds the source matrix from the tick proxy's current window and reuses calculate for exact parity with the backtest. No-op when the proxy is empty.

Source code in src/tradetropy/ta/order_flow/large_trades.py

def live_refresh(self, proxy) -> None:
    """
    Re-run causal detection over the source tick window (live mode).

    Tick-mounted indicators are not recomputed by the standard live indicator
    path (which reads OHLC ring bands), so the generic live primitive updater
    calls this hook before draw(): it rebuilds the source matrix from the
    tick proxy's current window and reuses calculate for exact parity with
    the backtest. No-op when the proxy is empty.
    """
    if proxy is None or len(proxy) == 0:
        return
    try:
        ts = np.asarray(proxy.ts[:], dtype=np.float64)
        price = np.asarray(proxy.price[:], dtype=np.float64)
        volume = np.asarray(proxy.volume[:], dtype=np.float64)
        flags = np.asarray(proxy.flags[:], dtype=np.float64)
        bid = np.asarray(proxy.bid[:], dtype=np.float64)
        ask = np.asarray(proxy.ask[:], dtype=np.float64)
    except Exception:
        return
    if ts.size == 0:
        return
    self.calculate(np.column_stack([ts, price, volume, flags, bid, ask]))
    self._accumulate_live_bubbles(ts)

CandlePatterns ¶

CandlePatterns(*, window: int = 100, doji_body='p10', small_body='p50', long_body='p60', wick='p80', marubozu_wick='p20', doji_frac: float = 0.1, zscore_len: int = 50, zscore_thresh: float = 0.5, context_filter: bool = True, horizon: int = 10, show_stats: bool = False, bull_color: str = _CANDLE_BULL_COLOR, bear_color: str = _CANDLE_BEAR_COLOR, neutral_color: str = _CANDLE_NEUTRAL_COLOR, label_offset: float = 0.01)

Bases: Indicator

Statistical candlestick pattern detector (annotation) with adaptive percentile thresholds, a z-score context filter and causal efficacy tracking.

Unlike a "detect anything" pattern list, the classification is relative to the recent distribution (a hammer needs a lower wick beyond the wick percentile of the last window bars and a small body) and, optionally, to the price context (a hammer only counts when price is stretched down, z-score of close vs its rolling mean <= -zscore_thresh). All detection is pure and causal (see tradetropy.ta._candlestick), so it is identical in backtest, live and replay.

It draws the pattern name on each candle (below bullish, above bearish) and exposes a public query API through the handle add_indicator returns, so a strategy reads the current pattern - and each pattern's empirical hit-rate - causally from on_data().

Usage

def init(self): self.btc = self.subscribe_ohlc("BTCUSDT", "1h", window_size=300) # Proxy-direct (source_cols resolves ts+OHLC), or CandlePatterns.refs(self.btc). self.candles = self.add_indicator(self.btc, CandlePatterns())

def on_data(self): if self.candles.last_pattern() == "Bullish Engulfing": # Only act if it has historically worked on THIS symbol/timeframe. eff = self.candles.efficacy("Bullish Engulfing") if eff["sample_size"] >= 20 and eff["hit_rate"] > 0.55: self.sesh.buy("BTCUSDT", volume=1)

Output bands (data, read in on_data via the handle): code[-1] -> numeric pattern code of the current bar (0 = none) direction[-1] -> +1 bullish / -1 bearish / 0 neutral

Query API (through the add_indicator() handle): last_pattern() -> name of the current bar's pattern ('none') pattern_at(i) -> name at offset i (e.g. -1 last, -2 previous) patterns(n=None) -> list of names (last n, or all in the window) is_bullish() / is_bearish() -> bias of the current bar name_of(code) -> label for a numeric code efficacy(pattern) -> {'hit_rate', 'sample_size', 'wins', 'name'} for one pattern (name or code), causal efficacy_all() -> {name: {...}} for every pattern seen so far

Parameters:

Name	Type	Description	Default
`window`	`int`	Trailing window for the adaptive size percentiles.	`100`
`doji_body/small_body/long_body/wick/marubozu_wick`	`str \| float`	Threshold specs (absolute, 'pXX' quantile or 'Nx' median multiple) - see `detect_candle_patterns`.	required
`doji_frac`	`float`	Max body/range for a doji.	`0.1`
`zscore_len`	`int`	Rolling window for the context z-score.	`50`
`zscore_thresh`	`float`	\|z\| beyond which price counts as stretched.	`0.5`
`context_filter`	`bool`	Gate hammer/hanging-man/inverted-hammer/shooting -star by context (default True).	`True`
`horizon`	`int`	Bars ahead used by the efficacy hit-rate.	`10`
`show_stats`	`bool`	Append each pattern's causal hit-rate to its drawn label (default False; the full dashboard is `efficacy_all()`).	`False`
`bull_color/bear_color/neutral_color`	`str`	Label colors by direction.	required
`label_offset`	`float`	Label distance from the wick, in percent of price.	`0.01`

Source code in src/tradetropy/ta/annotations.py

def __init__(
    self,
    *,
    window: int = 100,
    doji_body="p10",
    small_body="p50",
    long_body="p60",
    wick="p80",
    marubozu_wick="p20",
    doji_frac: float = 0.1,
    zscore_len: int = 50,
    zscore_thresh: float = 0.5,
    context_filter: bool = True,
    horizon: int = 10,
    show_stats: bool = False,
    bull_color: str = _CANDLE_BULL_COLOR,
    bear_color: str = _CANDLE_BEAR_COLOR,
    neutral_color: str = _CANDLE_NEUTRAL_COLOR,
    label_offset: float = 0.01,
):
    self.window = int(window)
    self.doji_body = doji_body
    self.small_body = small_body
    self.long_body = long_body
    self.wick = wick
    self.marubozu_wick = marubozu_wick
    self.doji_frac = float(doji_frac)
    self.zscore_len = int(zscore_len)
    self.zscore_thresh = float(zscore_thresh)
    self.context_filter = bool(context_filter)
    self.horizon = int(horizon)
    self.show_stats = bool(show_stats)
    self._bull_color = bull_color
    self._bear_color = bear_color
    self._neutral_color = neutral_color
    self._label_offset = float(label_offset) / 100.0

    # Full-series arrays stored by calculate() for draw().
    self._ts = None
    self._open = self._high = self._low = self._close = None
    self._codes = None
    self._directions = None
    # Source proxy for causal on-demand recomputation (set by add_indicator).
    self._src_proxy = None

    self.plot_config = IndicatorPlotConfig(
        overlay=True,
        exclude_from_autoscale=True,
        renderer="none",
        plot=True,
        name="Candles",
    )

calculate ¶

calculate(source: ndarray) -> np.ndarray

Parameters:

Name	Type	Description	Default
`source`	`ndarray`	[N x 5] - ts(0), open(1), high(2), low(3), close(4)	required

Returns:

Type	Description
`ndarray`	[2 x N] - row 0 = pattern code (0 = none), row 1 = direction bias.

Source code in src/tradetropy/ta/annotations.py

def calculate(self, source: np.ndarray) -> np.ndarray:
    """
    Args:
        source: [N x 5] - ts(0), open(1), high(2), low(3), close(4)

    Returns:
        [2 x N] - row 0 = pattern code (0 = none), row 1 = direction bias.
    """
    from tradetropy.ta._candlestick import detect_candle_patterns

    n = len(source)
    if n == 0 or source.ndim != 2 or source.shape[1] < 5:
        return np.zeros((2, max(n, 1)), dtype=np.float64)

    ts = source[:, 0].astype(np.float64)
    open_ = source[:, 1].astype(np.float64)
    high = source[:, 2].astype(np.float64)
    low = source[:, 3].astype(np.float64)
    close = source[:, 4].astype(np.float64)

    codes, directions = detect_candle_patterns(
        open_, high, low, close, **self._detect_kwargs()
    )

    self._ts = ts
    self._open, self._high, self._low, self._close = open_, high, low, close
    self._codes = codes
    self._directions = directions

    out = np.zeros((2, n), dtype=np.float64)
    out[0] = codes.astype(np.float64)
    out[1] = directions
    return out

set_query_source ¶

set_query_source(proxy) -> None

Wire the source OHLC proxy for causal on-demand recomputation.

Source code in src/tradetropy/ta/annotations.py

def set_query_source(self, proxy) -> None:
    """Wire the source OHLC proxy for causal on-demand recomputation."""
    self._src_proxy = proxy

last_pattern ¶

last_pattern() -> str

Name of the current (last) bar's pattern ('none' if there is none).

Source code in src/tradetropy/ta/annotations.py

def last_pattern(self) -> str:
    """Name of the current (last) bar's pattern ('none' if there is none)."""
    from tradetropy.ta._candlestick import pattern_name

    codes, _, _ = self._causal_codes()
    if len(codes) == 0:
        return "none"
    return pattern_name(int(codes[-1]))

pattern_at ¶

pattern_at(i: int) -> str

Name of the pattern at offset i within the causal window.

Parameters:

Name	Type	Description	Default
`i`	`int`	Index (e.g. -1 current bar, -2 previous). Out of range returns 'none'.	required

Source code in src/tradetropy/ta/annotations.py

def pattern_at(self, i: int) -> str:
    """
    Name of the pattern at offset ``i`` within the causal window.

    Args:
        i (int): Index (e.g. -1 current bar, -2 previous). Out of range
            returns 'none'.
    """
    from tradetropy.ta._candlestick import pattern_name

    codes, _, _ = self._causal_codes()
    n = len(codes)
    if n == 0 or i < -n or i >= n:
        return "none"
    return pattern_name(int(codes[i]))

patterns ¶

patterns(n: int | None = None) -> list

List of pattern names over the causal window.

Parameters:

Name	Type	Description	Default
`n`	`int \| None`	Return only the last `n` bars (None -> all).	`None`

Source code in src/tradetropy/ta/annotations.py

def patterns(self, n: "int | None" = None) -> list:
    """
    List of pattern names over the causal window.

    Args:
        n (int | None): Return only the last ``n`` bars (None -> all).
    """
    from tradetropy.ta._candlestick import pattern_name

    codes, _, _ = self._causal_codes()
    names = [pattern_name(int(x)) for x in codes]
    return names[-n:] if n else names

is_bullish ¶

is_bullish() -> bool

True if the current bar's pattern has a bullish bias.

Source code in src/tradetropy/ta/annotations.py

def is_bullish(self) -> bool:
    """True if the current bar's pattern has a bullish bias."""
    _, directions, _ = self._causal_codes()
    return len(directions) > 0 and directions[-1] > 0

is_bearish ¶

is_bearish() -> bool

True if the current bar's pattern has a bearish bias.

Source code in src/tradetropy/ta/annotations.py

def is_bearish(self) -> bool:
    """True if the current bar's pattern has a bearish bias."""
    _, directions, _ = self._causal_codes()
    return len(directions) > 0 and directions[-1] < 0

name_of ¶

name_of(code) -> str

Human-readable label for a numeric pattern code.

Source code in src/tradetropy/ta/annotations.py

def name_of(self, code) -> str:
    """Human-readable label for a numeric pattern code."""
    from tradetropy.ta._candlestick import pattern_name

    return pattern_name(code)

efficacy ¶

efficacy(pattern) -> dict

Causal empirical hit-rate for one pattern (name or code).

Only signals resolved by the current bar (s + horizon bars elapsed) are counted, so the statistic never uses future information.

Returns:

Name	Type	Description
`dict`	`dict`	{'hit_rate', 'sample_size', 'wins', 'name'}. An unseen pattern yields sample_size 0 and hit_rate NaN.

Source code in src/tradetropy/ta/annotations.py

def efficacy(self, pattern) -> dict:
    """
    Causal empirical hit-rate for one pattern (name or code).

    Only signals resolved by the current bar (``s + horizon`` bars elapsed)
    are counted, so the statistic never uses future information.

    Returns:
        dict: {'hit_rate', 'sample_size', 'wins', 'name'}. An unseen pattern
            yields sample_size 0 and hit_rate NaN.
    """
    from tradetropy.ta._candlestick import evaluate_efficacy, pattern_name

    code = self._resolve_code(pattern)
    codes, directions, close = self._causal_codes()
    stats = evaluate_efficacy(codes, directions, close, horizon=self.horizon)
    if code is not None and code in stats:
        return stats[code]
    return {
        "hit_rate": float("nan"),
        "sample_size": 0,
        "wins": 0,
        "name": pattern_name(code) if code is not None else str(pattern),
    }

efficacy_all ¶

efficacy_all() -> dict

Causal empirical hit-rate for every pattern seen so far, keyed by name.

Returns:

Type	Description
`dict`	dict[str, dict]: name -> {'hit_rate', 'sample_size', 'wins', 'code'}.

Source code in src/tradetropy/ta/annotations.py

def efficacy_all(self) -> dict:
    """
    Causal empirical hit-rate for every pattern seen so far, keyed by name.

    Returns:
        dict[str, dict]: name -> {'hit_rate', 'sample_size', 'wins', 'code'}.
    """
    from tradetropy.ta._candlestick import evaluate_efficacy

    codes, directions, close = self._causal_codes()
    stats = evaluate_efficacy(codes, directions, close, horizon=self.horizon)
    out = {}
    for code, entry in stats.items():
        out[entry["name"]] = {
            "hit_rate": entry["hit_rate"],
            "sample_size": entry["sample_size"],
            "wins": entry["wins"],
            "code": code,
        }
    return out

draw ¶

draw(cfg=None, *, interval_ms=None) -> list

Emit the pattern name over each detected candle: below bullish candles, above bearish/neutral candles. With show_stats the pattern's causal hit-rate (as of the last bar) is appended to the label.

Source code in src/tradetropy/ta/annotations.py

def draw(self, cfg=None, *, interval_ms=None) -> list:
    """
    Emit the pattern name over each detected candle: below bullish candles,
    above bearish/neutral candles. With ``show_stats`` the pattern's causal
    hit-rate (as of the last bar) is appended to the label.
    """
    from tradetropy.ta.draw import Labels
    from tradetropy.ta._candlestick import (
        pattern_name, PATTERN_DIRECTION, evaluate_efficacy,
    )

    if self._codes is None or len(self._codes) == 0:
        return []

    eff = {}
    if self.show_stats:
        stats = evaluate_efficacy(
            self._codes, self._directions, self._close, horizon=self.horizon
        )
        eff = {c: e["hit_rate"] for c, e in stats.items()}

    bull_x, bull_y, bull_txt = [], [], []
    top_x, top_y, top_txt, top_color = [], [], [], []
    for i in range(len(self._codes)):
        code = int(self._codes[i])
        if code == 0:
            continue
        label = pattern_name(code)
        if self.show_stats and code in eff and not np.isnan(eff[code]):
            label = f"{label} {eff[code]:.0%}"
        direction = PATTERN_DIRECTION.get(code, 0.0)
        ts_i = int(self._ts[i])
        if direction > 0:
            bull_x.append(ts_i)
            bull_y.append(float(self._low[i]) * (1.0 - self._label_offset))
            bull_txt.append(label)
        else:
            top_x.append(ts_i)
            top_y.append(float(self._high[i]) * (1.0 + self._label_offset))
            top_txt.append(label)
            top_color.append(
                self._bear_color if direction < 0 else self._neutral_color
            )

    prims = []
    if bull_x:
        prims.append(Labels(
            x=bull_x, y=bull_y, text=bull_txt, color=self._bull_color,
            font_size="8pt", x_offset=0, y_offset=0,
            text_align="center", text_baseline="top",
        ))
    if top_x:
        prims.append(Labels(
            x=top_x, y=top_y, text=top_txt, color=top_color,
            font_size="8pt", x_offset=0, y_offset=0,
            text_align="center", text_baseline="bottom",
        ))
    return prims

ManualMarks ¶

ManualMarks(max_marks: int = 500)

Bases: Indicator

Strategy-driven manual marks: segments with a start and an end that the strategy adds/updates/removes from on_data().

A mark is a line in price x time space defined by two endpoints (ts0, price0) -> (ts1, price1). Use the same price for both endpoints for a horizontal level; use different prices for a sloped line. ts1 (and price1) can be left open (None) while the mark is "live" - draw() extends it to the current bar/tick until the strategy closes it.

This indicator draws no series (its clock band is a NaN placeholder, required only so add_indicator() returns a query-capable proxy - see n_outputs note above the class attributes). Attach it to any already-subscribed source purely as a causal clock (ManualMarks.refs(proxy)), then reach it through the handle add_indicator() returns.

Usage

def init(self): self.btc = self.subscribe_ohlc('BTCUSDT', '5m') self.marks = self.add_indicator( ManualMarks.refs(self.btc), ManualMarks(), )

def on_data(self): if some_signal: self.mark_id = self.marks.add_mark( price0=self.btc.close[-1], ts0=self.ts, color='#F6465D', label='Signal', ) if close_condition and self.mark_id is not None: self.marks.close_mark(self.mark_id, ts1=self.ts, price1=self.btc.close[-1])

Query API (through the add_indicator() handle): add_mark(price0, ts0=None, price1=None, ts1=None, color=None, dash='solid', width=1.6, label=None, marker=None) -> int Adds a mark and returns its id. ts0 defaults to the current clock (self.ts equivalent); price1/ts1 left None keep the mark open (draw() extends it to the latest known ts). marker (optional) also drops a point at (ts0, price0) - e.g. to flag the pivot that triggered the mark. update_mark(mark_id, **fields) -> None Updates any field of an existing mark (e.g. drag the open end). close_mark(mark_id, ts1=None, price1=None) -> None Fixes the end of an open mark. ts1 defaults to the current clock; price1 defaults to the mark's own price0 (horizontal level). remove_mark(mark_id) -> None Deletes a mark (no-op if the id does not exist). clear_marks() -> None Deletes every mark. marks -> list[dict] Read-only snapshot of the current marks (copies).

Parameters:

Name	Type	Description	Default
`max_marks`	`int`	Ring cap on stored marks (oldest dropped first, 0 = unlimited). Keeps a long-running live session bounded.	`500`

Source code in src/tradetropy/ta/manual_marks.py

def __init__(self, max_marks: int = 500):
    self.max_marks = int(max_marks)
    self._marks: dict[int, dict] = {}
    self._next_id = 1
    self._last_ts: int | None = None

    self.plot_config = IndicatorPlotConfig(
        overlay=True,
        exclude_from_autoscale=True,
        renderer='none',
        plot=True,
        name='Marks',
    )

marks `property` ¶

marks: list

Read-only snapshot of the current marks (list of dict copies).

refs `staticmethod` ¶

refs(proxy)

Build the ColumnRef list for this indicator: only a causal clock.

Parameters:

Name	Type	Description	Default
`proxy`	`OhlcProxy \| TickProxy`	Any already-subscribed source.	required

Returns:

Type	Description
	list[ColumnRef]: [ts] ref (close is not needed, kept minimal).

Source code in src/tradetropy/ta/manual_marks.py

@staticmethod
def refs(proxy):
    """
    Build the ColumnRef list for this indicator: only a causal clock.

    Args:
        proxy (OhlcProxy | TickProxy): Any already-subscribed source.

    Returns:
        list[ColumnRef]: [ts] ref (close is not needed, kept minimal).
    """
    return [proxy.ts_ref]

add_mark ¶

add_mark(price0: float, ts0: 'int | None' = None, price1: 'float | None' = None, ts1: 'int | None' = None, *, color: 'str | None' = None, dash: str = 'solid', width: float = 1.6, alpha: float = 0.9, label: 'str | None' = None, marker: 'str | None' = None) -> int

Add a manual mark and return its id.

Parameters:

Name	Type	Description	Default
`price0`	`float`	Start price.	required
`ts0`	`int \| None`	Start timestamp (epoch ms). None -> current causal clock (the ts of the last processed bar/tick).	`None`
`price1`	`float \| None`	End price. None -> mark stays open (horizontal at price0 until closed or drawn to "now").	`None`
`ts1`	`int \| None`	End timestamp. None -> mark stays open.	`None`
`color`	`str \| None`	Line/point color. None -> module default.	`None`
`dash`	`str`	Line style ('solid'\|'dashed'\|'dotted'\|'dashdot'\|'dotdash').	`'solid'`
`width`	`float`	Line width.	`1.6`
`alpha`	`float`	Line opacity.	`0.9`
`label`	`str \| None`	Optional floating text at the start point.	`None`
`marker`	`str \| None`	Optional marker shape dropped at (ts0, price0).	`None`

Returns:

Name	Type	Description
`int`	`int`	The new mark's id (pass it to update_mark/close_mark/remove_mark).

Source code in src/tradetropy/ta/manual_marks.py

def add_mark(
    self,
    price0: float,
    ts0: "int | None" = None,
    price1: "float | None" = None,
    ts1: "int | None" = None,
    *,
    color: "str | None" = None,
    dash: str = 'solid',
    width: float = 1.6,
    alpha: float = 0.9,
    label: "str | None" = None,
    marker: "str | None" = None,
) -> int:
    """
    Add a manual mark and return its id.

    Args:
        price0 (float): Start price.
        ts0 (int | None): Start timestamp (epoch ms). None -> current
            causal clock (the ts of the last processed bar/tick).
        price1 (float | None): End price. None -> mark stays open
            (horizontal at price0 until closed or drawn to "now").
        ts1 (int | None): End timestamp. None -> mark stays open.
        color (str | None): Line/point color. None -> module default.
        dash (str): Line style ('solid'|'dashed'|'dotted'|'dashdot'|'dotdash').
        width (float): Line width.
        alpha (float): Line opacity.
        label (str | None): Optional floating text at the start point.
        marker (str | None): Optional marker shape dropped at (ts0, price0).

    Returns:
        int: The new mark's id (pass it to update_mark/close_mark/remove_mark).
    """
    ts0 = int(ts0) if ts0 is not None else self._last_ts
    if ts0 is None:
        raise ConfigError(
            'ManualMarks.add_mark(): ts0 is required until the first bar/'
            'tick has been processed (no causal clock yet).'
        )
    mark_id = self._next_id
    self._next_id += 1
    self._marks[mark_id] = {
        'id': mark_id,
        'ts0': ts0,
        'price0': float(price0),
        'ts1': int(ts1) if ts1 is not None else None,
        'price1': float(price1) if price1 is not None else None,
        'color': color or _DEFAULT_COLOR,
        'dash': dash,
        'width': float(width),
        'alpha': float(alpha),
        'label': label,
        'marker': marker,
    }
    if self.max_marks and len(self._marks) > self.max_marks:
        oldest_id = next(iter(self._marks))
        del self._marks[oldest_id]
    return mark_id

update_mark ¶

update_mark(mark_id: int, **fields) -> None

Update fields of an existing mark (e.g. drag the open end as price moves). Unknown mark_id is a no-op.

Parameters:

Name	Type	Description	Default
`mark_id`	`int`	Id returned by add_mark().	required
`**fields`		Any of price0/ts0/price1/ts1/color/dash/width/alpha/ label/marker.	`{}`

Source code in src/tradetropy/ta/manual_marks.py

def update_mark(self, mark_id: int, **fields) -> None:
    """
    Update fields of an existing mark (e.g. drag the open end as price
    moves). Unknown mark_id is a no-op.

    Args:
        mark_id (int): Id returned by add_mark().
        **fields: Any of price0/ts0/price1/ts1/color/dash/width/alpha/
            label/marker.
    """
    mark = self._marks.get(mark_id)
    if mark is None:
        return
    valid = {'price0', 'ts0', 'price1', 'ts1', 'color', 'dash', 'width',
             'alpha', 'label', 'marker'}
    bad = set(fields) - valid
    if bad:
        raise ConfigError(f'ManualMarks.update_mark(): unknown fields {bad}')
    for key, value in fields.items():
        if key in ('ts0', 'ts1') and value is not None:
            value = int(value)
        elif key in ('price0', 'price1') and value is not None:
            value = float(value)
        mark[key] = value

close_mark ¶

close_mark(mark_id: int, ts1: 'int | None' = None, price1: 'float | None' = None) -> None

Fix the end of an open mark.

Parameters:

Name	Type	Description	Default
`mark_id`	`int`	Id returned by add_mark().	required
`ts1`	`int \| None`	End timestamp. None -> current causal clock.	`None`
`price1`	`float \| None`	End price. None -> the mark's own price0 (horizontal level).	`None`

Source code in src/tradetropy/ta/manual_marks.py

def close_mark(
    self,
    mark_id: int,
    ts1: "int | None" = None,
    price1: "float | None" = None,
) -> None:
    """
    Fix the end of an open mark.

    Args:
        mark_id (int): Id returned by add_mark().
        ts1 (int | None): End timestamp. None -> current causal clock.
        price1 (float | None): End price. None -> the mark's own price0
            (horizontal level).
    """
    mark = self._marks.get(mark_id)
    if mark is None:
        return
    mark['ts1'] = int(ts1) if ts1 is not None else self._last_ts
    mark['price1'] = float(price1) if price1 is not None else mark['price0']

remove_mark ¶

remove_mark(mark_id: int) -> None

Delete a mark. No-op if mark_id does not exist.

Source code in src/tradetropy/ta/manual_marks.py

def remove_mark(self, mark_id: int) -> None:
    """Delete a mark. No-op if mark_id does not exist."""
    self._marks.pop(mark_id, None)

clear_marks ¶

clear_marks() -> None

Delete every mark.

Source code in src/tradetropy/ta/manual_marks.py

def clear_marks(self) -> None:
    """Delete every mark."""
    self._marks.clear()

draw ¶

draw(cfg=None, *, interval_ms=None) -> list

Emit the current marks as Segments (sloped/horizontal lines), plus optional Points (markers) and Labels for marks that requested them.

Open marks (price1/ts1 is None) are extended to the latest causal clock so a live/backtest mark is visible as soon as it is added, without waiting for it to be closed.

Source code in src/tradetropy/ta/manual_marks.py

def draw(self, cfg=None, *, interval_ms=None) -> list:
    """
    Emit the current marks as Segments (sloped/horizontal lines), plus
    optional Points (markers) and Labels for marks that requested them.

    Open marks (price1/ts1 is None) are extended to the latest causal
    clock so a live/backtest mark is visible as soon as it is added,
    without waiting for it to be closed.
    """
    from tradetropy.ta.draw import Segments, Points, Labels

    if not self._marks:
        return []

    now_ts = self._last_ts
    # Segments.width/dash are scalar per primitive (color/alpha are per-
    # element), so marks are grouped by (dash, width) into one Segments
    # primitive per group instead of losing per-mark line style.
    groups: dict[tuple, dict] = {}
    pt_x, pt_y, pt_color, pt_marker_groups = [], [], [], {}
    lbl_x, lbl_y, lbl_text, lbl_color = [], [], [], []

    for mark in self._marks.values():
        ts0 = mark['ts0']
        price0 = mark['price0']
        ts1 = mark['ts1'] if mark['ts1'] is not None else (now_ts if now_ts is not None else ts0)
        price1 = mark['price1'] if mark['price1'] is not None else price0
        if ts1 < ts0:
            ts0, ts1 = ts1, ts0
            price0, price1 = price1, price0

        key = (mark['dash'], mark['width'])
        grp = groups.setdefault(key, {'x0': [], 'y0': [], 'x1': [], 'y1': [],
                                       'color': [], 'alpha': []})
        grp['x0'].append(int(ts0)); grp['y0'].append(float(price0))
        grp['x1'].append(int(ts1)); grp['y1'].append(float(price1))
        grp['color'].append(mark['color'])
        grp['alpha'].append(mark['alpha'])

        if mark['marker']:
            pt_x.append(int(mark['ts0'])); pt_y.append(float(mark['price0']))
            pt_color.append(mark['color'])
            pt_marker_groups.setdefault(mark['marker'], []).append(len(pt_x) - 1)

        if mark['label']:
            lbl_x.append(int(mark['ts0'])); lbl_y.append(float(mark['price0']))
            lbl_text.append(mark['label'])
            lbl_color.append(mark['color'])

    prims: list = []
    for (dash, width), grp in groups.items():
        prims.append(Segments(
            x0=grp['x0'], y0=grp['y0'], x1=grp['x1'], y1=grp['y1'],
            color=grp['color'], alpha=grp['alpha'], width=width, dash=dash,
        ))
    # Points.marker is also scalar per primitive: one Points group per
    # distinct marker shape.
    for marker_shape, idxs in pt_marker_groups.items():
        prims.append(Points(
            x=[pt_x[i] for i in idxs], y=[pt_y[i] for i in idxs],
            color=[pt_color[i] for i in idxs], marker=marker_shape,
        ))
    if lbl_x:
        prims.append(Labels(
            x=lbl_x, y=lbl_y, text=lbl_text, color=lbl_color,
            font_size='8pt', x_offset=4, y_offset=2,
            text_align='left', text_baseline='middle',
        ))
    return prims

Indicators¶

ta ¶

Indicator ¶

CLASS ATTRIBUTES (override in subclasses)¶

METHODS¶

plot_config property writable ¶

calculate ¶

draw ¶

refs classmethod ¶

default_refs ¶

col_name ¶

display_name ¶

IndicatorPlotConfig dataclass ¶

Visibility fields¶

Line fields (when scatter=False) - scalar or list per dimension¶

Scatter fields (when scatter=True) - scalar or list per dimension¶

Reference lines¶

Per-dimension style usage¶

scatter property ¶

resolve_overlay ¶

dim ¶

SMA ¶

EMA ¶

MACD ¶

RSI ¶

calculate ¶

ATR ¶

BollingerBands ¶

VolumeProfile ¶

refs staticmethod ¶

RollingVolumeProfile ¶

period_profiles ¶

LargeTrades ¶

refs staticmethod ¶

default_refs ¶

detect ¶

calculate ¶

event_labels ¶

draw ¶

live_refresh ¶

CandlePatterns ¶

calculate ¶

set_query_source ¶

last_pattern ¶

pattern_at ¶

patterns ¶

is_bullish ¶

is_bearish ¶

name_of ¶

efficacy ¶

efficacy_all ¶

draw ¶

ManualMarks ¶

marks property ¶

refs staticmethod ¶

add_mark ¶

update_mark ¶

close_mark ¶

remove_mark ¶

clear_marks ¶

draw ¶

plot_config `property` `writable` ¶

refs `classmethod` ¶

IndicatorPlotConfig `dataclass` ¶

scatter `property` ¶

refs `staticmethod` ¶

refs `staticmethod` ¶

marks `property` ¶

refs `staticmethod` ¶