TradePose Models

Shared Pydantic models, enums, and schemas for all TradePose packages.

Features

• ✅ Strategy Configuration - StrategyConfig, Blueprint, Trigger, IndicatorSpec
• ✅ Registry & Portfolio - Multi-strategy management with unique key indexing
• ✅ Enumerations - Freq, TradeDirection, OrderStrategy, EngagementPhase, ActionType
• ✅ Polars Schemas - DataFrame schemas for trades and OHLCV
• ✅ Type Safety - Full Pydantic validation with IDE autocomplete

Installation

pip install tradepose-models

Requirements:

• Python 3.13+
• Dependencies: pydantic, polars

Quick Start

Strategy Configuration

from tradepose_models.strategy import (
    StrategyConfig,
    Blueprint,
    Trigger,
    IndicatorSpec,
)
from tradepose_models.enums import (
    Freq,
    TradeDirection,
    TrendType,
    OrderStrategy,
)
from tradepose_models.indicators import Indicator
import polars as pl

# Create indicator
atr_spec = IndicatorSpec(
    freq=Freq.DAY_1,
    shift=1,
    indicator=Indicator.atr(period=14),
)

# Create trigger
entry_trigger = Trigger(
    name="breakout_entry",
    order_strategy=OrderStrategy.IMMEDIATE_ENTRY,
    priority=1,
    conditions=[pl.col("close") > pl.col("high").shift(1)],
    offset_expr=pl.lit(0.0),
)

# Create blueprint
blueprint = Blueprint(
    name="trend_long",
    direction=TradeDirection.LONG,
    trend_type=TrendType.TREND,
    entry_triggers=[entry_trigger],
    exit_triggers=[...],
)

# Create strategy
strategy = StrategyConfig(
    name="ES_Breakout",
    base_instrument="ES",
    base_freq=Freq.MIN_15,
    note="E-mini S&P breakout strategy",
    indicators=[atr_spec],
    base_blueprint=blueprint,
)

# Save/Load
strategy.save("strategy.json")
loaded = StrategyConfig.load("strategy.json")

Strategy Registry

Manage multiple strategies with (strategy_name, blueprint_name) unique keys:

from tradepose_models.strategy import StrategyRegistry, StrategyConfig

# Create registry
registry = StrategyRegistry()

# Add strategy (auto-splits blueprints into separate entries)
strategy = StrategyConfig.load("multi_blueprint_strategy.json")
keys = registry.add(strategy)
# keys = [(BlueprintSelection("MyStrategy", "long_bp"), BlueprintSelection("MyStrategy", "short_bp")]

# Access entries
entry = registry.get("MyStrategy", "long_bp")
print(entry.blueprint.direction)  # TradeDirection.LONG

# Get all configs for backtesting
configs = registry.get_configs()  # List[StrategyConfig] with single blueprints

Portfolio Selection

Create portfolios by selecting from registry:

from tradepose_models.strategy import StrategyRegistry

registry = StrategyRegistry()
registry.add(strategy1)  # 2 blueprints
registry.add(strategy2)  # 3 blueprints

# Create portfolio (selection view)
portfolio = registry.select(
    name="Q1_Momentum",
    selections=[
        ("Strategy1", "momentum_long"),
        ("Strategy1", "momentum_short"),
        ("Strategy2", "trend_follow"),
    ],
    capital=100000,
    currency="USD",
    platform="MT5",
)

# Get configs for BatchTester
configs = portfolio.get_configs(registry)

# Use with BatchTester
from tradepose_client import BatchTester
from tradepose_client.batch import Period

tester = BatchTester(api_key="tp_xxx")
batch = tester.submit_backtest(strategies=configs, periods=[Period.Q1(2024)])
batch.wait()

Core Concepts

Strategy Hierarchy

StrategyConfig
├── name, base_instrument, base_freq
├── indicators: List[IndicatorSpec]
├── base_blueprint: Blueprint
│   ├── name, direction, trend_type
│   ├── entry_triggers: List[Trigger]
│   │   └── conditions: List[pl.Expr], price_expr: pl.Expr
│   └── exit_triggers: List[Trigger]
└── advanced_blueprints: List[Blueprint]

Registry & Portfolio

StrategyRegistry (storage, guarantees uniqueness)
│
│  Key: (strategy_name, blueprint_name)
│
├── add(config) → auto-split blueprints
├── get(name, bp) → RegistryEntry
├── get_configs() → List[StrategyConfig]  ──> for BatchTester
└── select(...) → Portfolio
                     │
                     ▼
              Portfolio (selection view)
              ├── selections: List[(name, bp)]
              ├── capital, currency, platform
              └── get_configs(registry) → configs

API Reference

Enumerations

Freq (Time Frequency)

from tradepose_models.enums import Freq

Freq.MIN_1   # "1min"
Freq.MIN_5   # "5min"
Freq.MIN_15  # "15min"
Freq.MIN_30  # "30min"
Freq.HOUR_1  # "1h"
Freq.HOUR_4  # "4h"
Freq.DAY_1   # "1D"
Freq.WEEK_1  # "1W"
Freq.MONTH_1 # "1M"

TradeDirection

from tradepose_models.enums import TradeDirection

TradeDirection.LONG   # Long trades only
TradeDirection.SHORT  # Short trades only

TrendType

from tradepose_models.enums import TrendType

TrendType.TREND     # Trend-following
TrendType.RANGE     # Mean-reversion
TrendType.REVERSAL  # Counter-trend

OrderStrategy

from tradepose_models.enums import OrderStrategy

# Entry
OrderStrategy.IMMEDIATE_ENTRY      # Execute immediately
OrderStrategy.FAVORABLE_DELAY_ENTRY  # Wait for pullback
OrderStrategy.ADVERSE_DELAY_ENTRY   # Wait for breakout

# Exit
OrderStrategy.IMMEDIATE_EXIT  # Exit immediately
OrderStrategy.STOP_LOSS       # Fixed stop loss
OrderStrategy.TAKE_PROFIT     # Fixed profit target
OrderStrategy.TRAILING_STOP   # Dynamic trailing
OrderStrategy.BREAKEVEN       # Move to breakeven
OrderStrategy.TIMEOUT_EXIT    # Time-based exit

Currency

from tradepose_models.enums import Currency

Currency.USD, Currency.USDT, Currency.TWD
Currency.EUR, Currency.JPY
Currency.BTC, Currency.ETH
Currency.XAU, Currency.TAIEX

Platform

from tradepose_models.enums import Platform

Platform.MT5      # MetaTrader 5
Platform.BINANCE  # Binance
Platform.SHIOAJI  # Shioaji (Taiwan)
Platform.CCXT     # CCXT unified

AccountSource

from tradepose_models.enums import AccountSource

AccountSource.FTMO        # tz_offset: 2
AccountSource.IB          # tz_offset: 0
AccountSource.FIVEPERCENT # tz_offset: 2
AccountSource.BINANCE     # tz_offset: 8
AccountSource.SHIOAJI     # tz_offset: 8

# Get timezone offset
offset = AccountSource.FTMO.tz_offset()  # 2

EngagementPhase (v0.3.0)

from tradepose_models.enums import EngagementPhase

# 9-state lifecycle
EngagementPhase.PENDING     # 0: Awaiting entry
EngagementPhase.ENTERING    # 1: Entry order placed
EngagementPhase.HOLDING     # 2: Position held
EngagementPhase.EXITING     # 3: Exit order placed
EngagementPhase.CLOSED      # 4: Position closed
EngagementPhase.FAILED      # 5: Entry failed
EngagementPhase.CANCELLED   # 6: Entry cancelled
EngagementPhase.EXIT_FAILED # 7: Exit failed
EngagementPhase.EXPIRED     # 8: Entry expired

ActionType (v0.3.0)

from tradepose_models.enums import ActionType

# BrokerExecutor dispatch
ActionType.NONE            # No action needed
ActionType.MARKET_ENTRY    # execute_market_entry()
ActionType.LIMIT_ENTRY     # execute_limit_entry()
ActionType.CLOSE_POSITION  # execute_close_position()
ActionType.MODIFY_POSITION # execute_modify_position()
ActionType.MODIFY_ORDER    # execute_modify_order()
ActionType.CANCEL_ORDER    # execute_cancel_order()

Strategy Models

StrategyConfig

from tradepose_models.strategy import StrategyConfig

# Fields
strategy.name              # str: Strategy name
strategy.base_instrument   # str: Trading instrument
strategy.base_freq         # Freq: Base timeframe
strategy.note              # str: Description
strategy.indicators        # List[IndicatorSpec]
strategy.base_blueprint    # Blueprint: Primary blueprint
strategy.advanced_blueprints  # List[Blueprint]: Additional blueprints

# Methods
StrategyConfig.load(path)     # Load from JSON file
strategy.save(path)           # Save to JSON file
strategy.to_json()            # Serialize to JSON string
strategy.to_dict()            # Convert to dict
StrategyConfig.from_api(data) # Parse from API response

Blueprint

from tradepose_models.strategy import Blueprint

# Fields
blueprint.name           # str: Blueprint name
blueprint.direction      # TradeDirection: LONG/SHORT
blueprint.trend_type     # TrendType: TREND/RANGE/REVERSAL
blueprint.entry_triggers # List[Trigger]
blueprint.exit_triggers  # List[Trigger]

Trigger

from tradepose_models.strategy import Trigger
import polars as pl

trigger = Trigger(
    name="breakout",
    order_strategy=OrderStrategy.IMMEDIATE_ENTRY,
    priority=1,
    conditions=[
        pl.col("close") > pl.col("high").shift(1),
        pl.col("volume") > pl.col("volume").rolling_mean(20),
    ],
    offset_expr=pl.lit(0.0),
)

# Access Polars expressions directly
expr = trigger.conditions[0]  # pl.Expr

IndicatorSpec

from tradepose_models.strategy import IndicatorSpec
from tradepose_models.indicators import Indicator

spec = IndicatorSpec(
    freq=Freq.DAY_1,
    shift=1,
    indicator=Indicator.atr(period=14),
)

# Properties
spec.short_name()    # "ATR|14"
spec.display_name()  # "1D_ATR|14_s1"
spec.col()           # pl.col("1D_ATR|14_s1")

Registry Models

StrategyRegistry

from tradepose_models.strategy import StrategyRegistry

registry = StrategyRegistry()

# Add strategy (auto-splits blueprints)
keys = registry.add(strategy)  # Returns List[BlueprintSelection]

# Add or replace
keys = registry.add_or_replace(strategy)

# Get entry
entry = registry.get("StrategyName", "blueprint_name")

# Get original strategy (with all blueprints)
original = registry.get_strategy("StrategyName")

# Get all configs (single-blueprint each)
configs = registry.get_configs()

# Remove strategy
removed_count = registry.remove("StrategyName")

# Create portfolio
portfolio = registry.select(
    name="MyPortfolio",
    selections=[("Strat1", "bp1"), ("Strat2", "bp2")],
    capital=100000,
)

# Iteration
for entry in registry:
    print(entry.key)

# Membership
if ("StrategyName", "blueprint") in registry:
    ...

# Length
print(f"Entries: {len(registry)}")

Portfolio

from tradepose_models.strategy import Portfolio, BlueprintSelection

# Create directly
portfolio = Portfolio(
    name="Q1_Portfolio",
    selections=[BlueprintSelection(strategy_name="Strat1", blueprint_name="bp1")],
    capital=100000,
    currency="USD",
    account_source="FTMO",
    platform="MT5",
)

# Or via registry.select()
portfolio = registry.select(...)

# Get configs for backtesting
configs = portfolio.get_configs(registry)

# Immutable operations (return new instance)
new_portfolio = portfolio.add_selection("Strat2", "bp2")
new_portfolio = portfolio.remove_selection("Strat1", "bp1")

# Properties
portfolio.strategy_names     # List[str] unique names
portfolio.selection_count    # int

# Serialization
portfolio.save("portfolio.json")
loaded = Portfolio.load("portfolio.json")
portfolio.to_json()
portfolio.to_dict()

Indicators

Indicator Factory

from tradepose_models.indicators import Indicator

# Moving Averages
Indicator.sma(period=20, column="close")
Indicator.ema(period=20, column="close")
Indicator.smma(period=20, column="close")
Indicator.wma(period=20, column="close")

# Volatility
Indicator.atr(period=14)
Indicator.atr_quantile(period=14, quantile=0.5, window=252)
Indicator.bollinger_bands(period=20, num_std=2.0)

# Trend
Indicator.supertrend(multiplier=3.0, volatility_column="ATR|14")
Indicator.macd(fast=12, slow=26, signal=9)
Indicator.adx(period=14)

# Momentum
Indicator.rsi(period=14)
Indicator.cci(period=20)
Indicator.stochastic(k_period=14, d_period=3)

# Volume Profile
Indicator.market_profile(
    period=30,
    tick_size=0.25,
    value_area_pct=0.70,
)

# Raw OHLCV
Indicator.raw_ohlcv(column="close")

Usage Examples

Multi-Strategy Backtesting

from tradepose_models.strategy import StrategyConfig, StrategyRegistry
from tradepose_client import BatchTester
from tradepose_client.batch import Period

# Setup registry
registry = StrategyRegistry()

# Load multiple strategies
for path in ["strat1.json", "strat2.json", "strat3.json"]:
    strategy = StrategyConfig.load(path)
    registry.add(strategy)

print(f"Registry: {len(registry)} entries")
# Registry: 8 entries (if strategies have multiple blueprints)

# Backtest all
configs = registry.get_configs()
tester = BatchTester(api_key="tp_xxx")
batch = tester.submit_backtest(
    strategies=configs,
    periods=[Period.Q1(2024), Period.Q2(2024)],
)
batch.wait()
print(batch.trades)

Portfolio-Based Testing

# Create focused portfolio
momentum_portfolio = registry.select(
    name="Momentum_Suite",
    selections=[
        ("TrendFollower", "momentum_long"),
        ("TrendFollower", "momentum_short"),
        ("BreakoutTrader", "range_breakout"),
    ],
    capital=50000,
    currency="USD",
)

# Test portfolio only
configs = momentum_portfolio.get_configs(registry)
batch = tester.submit_backtest(strategies=configs, periods=[Period.Q1(2024)])

# Save portfolio for later
momentum_portfolio.save("momentum_portfolio.json")

Indicator with Struct Fields

from tradepose_models.strategy import IndicatorSpec, Trigger
from tradepose_models.indicators import Indicator
from tradepose_models.enums import Freq, OrderStrategy
import polars as pl

# SuperTrend returns struct with multiple fields
supertrend = IndicatorSpec(
    freq=Freq.DAY_1,
    shift=1,
    indicator=Indicator.supertrend(multiplier=3.0, volatility_column="ATR|14"),
)

# Access struct fields in conditions
trigger = Trigger(
    name="supertrend_long",
    conditions=[
        supertrend.col().struct.field("direction") == 1,  # Long signal
        supertrend.col().struct.field("supertrend") < pl.col("close"),
    ],
    offset_expr=pl.lit(0.0),
    order_strategy=OrderStrategy.IMMEDIATE_ENTRY,
    priority=1,
)

Enhanced OHLCV Timezone Schema

Enhanced OHLCV is evaluated in the owning strategy's IANA timezone. Build its complete schema with that explicit runtime value:

from tradepose_models.schemas import (
    enhanced_ohlcv_cast_schema,
    get_enhanced_ohlcv_schema,
)

complete_schema = get_enhanced_ohlcv_schema(strategy.tz)
assert complete_schema["ts"] == pl.Datetime("ms", strategy.tz)

get_enhanced_ohlcv_schema() includes ts and is intended for complete-result construction or validation. enhanced_ohlcv_cast_schema intentionally excludes ts; use it only to normalize timezone-independent columns while preserving the timezone metadata already present after an engine result or Parquet read. There is no timezone-naive or hard-coded UTC default for Enhanced OHLCV.

Trades use Datetime("ms", "UTC") so multi-strategy results can be concatenated. Trade rows do not store strategy_tz; local display needs the original strategy_name -> IANA timezone mapping, and historical evaluation timezone cannot be recovered from a trade alone if that mapping is lost.

File Structure

packages/models/src/tradepose_models/
├── __init__.py
├── base.py                 # BaseModel configuration
├── enums/
│   ├── __init__.py
│   ├── freq.py             # Time frequencies
│   ├── trade_direction.py  # LONG/SHORT
│   ├── trend_type.py       # TREND/RANGE/REVERSAL
│   ├── order_strategy.py   # Entry/exit strategies
│   ├── currency.py         # Currency types
│   ├── platform.py         # Trading platforms
│   ├── account_source.py   # Brokers/prop firms
│   └── ...
├── strategy/
│   ├── __init__.py
│   ├── config.py           # StrategyConfig
│   ├── blueprint.py        # Blueprint
│   ├── trigger.py          # Trigger
│   ├── indicator_spec.py   # IndicatorSpec
│   ├── registry.py         # StrategyRegistry, RegistryEntry, BlueprintSelection
│   ├── portfolio.py        # Portfolio
│   └── helpers.py          # Factory functions
├── indicators/
│   ├── __init__.py
│   ├── factory.py          # Indicator factory
│   └── ...                 # Indicator type definitions
├── schemas/
│   ├── __init__.py
│   ├── trades.py           # Trades schema
│   ├── performance.py      # Performance schema
│   └── enhanced_ohlcv.py   # Strategy-local Enhanced OHLCV schema and cast map
└── ...

Development

Running Tests

# Run models tests
uv run python -m pytest tests/models/ -v

# Run with coverage
uv run python -m pytest tests/models/ --cov=tradepose_models

License

MIT License - see LICENSE file for details.

Support

• Documentation: docs/
• Issues: GitHub Issues