aptis-strategy-sdk 0.5.3

Python SDK for building plugin trading strategies on the Aptis platform

Aptis Strategy SDK

Python SDK for building institutional-grade trading strategies on the Aptis platform.

Author: Richard Chung
Version: 0.5.0
Python: 3.9+

---

5-Minute Quickstart

1. Install

pip install aptis-strategy-sdk

2. Configure

Create a .env file in your project directory:

APTIS_API_KEY=your_api_key_here
APTIS_API_URL=https://your-platform-url
APTIS_ACCOUNT=Theme

Never hardcode credentials. Always read from environment variables.

3. Write your first strategy

import os
from datetime import date
from aptis_strategy_sdk import AptisClient, Strategy, Signal

class MyStrategy(Strategy):
    def generate_signals(self, run_date: date):
        features = self.get_features(["AAPL", "MSFT"], run_date, ["rsi_14"])
        signals = []
        for symbol, data in features.items():
            if data.get("rsi_14", 50) < 30:
                signals.append(Signal(
                    symbol=symbol,
                    asset_class="equity",
                    quantity=100,
                    side="BUY",
                    signal_type="ENTRY",
                    metadata={"rsi": data["rsi_14"], "reason": "oversold"},
                ))
        return signals

client = AptisClient.from_env()
strategy = MyStrategy("My Strategy", client, os.getenv("APTIS_ACCOUNT", "Theme"))
result = strategy.run(date.today())
print(f"Submitted {result['submitted']} signal(s)")

4. Run in dry-run mode first

strategy = MyStrategy("My Strategy", client, "Theme", dry_run=True)
result = strategy.run(date.today())
# Signals are validated and logged — nothing is submitted

---

Core Concepts

Strategy

A strategy is a Python class that subclasses Strategy and implements generate_signals(run_date). The base class handles:

• Fetching config (NAV, funds, enabled flag) from the platform
• Skipping execution when the strategy is disabled
• Validating each signal before submission
• Logging submission results
• Calling lifecycle hooks (before_run, after_run, on_error)

class MyStrategy(Strategy):
    def generate_signals(self, run_date: date) -> list[Signal]:
        ...

Signal

A signal is an instruction to enter, exit, or adjust a position. It is the primary unit of communication between your strategy and the platform.

Signal → Platform → Trade record → Position update → P&L

Every signal requires five fields:

Field	Values	Description
symbol	e.g. "AAPL", "BTC/USD"	Ticker
asset_class	equity etf crypto forex commodity	Asset type
quantity	float > 0	Units to trade
side	BUY SELL	Direction
signal_type	ENTRY EXIT ADJUST	Intent

Position

A position is an open trade held by the strategy. Positions are created by ENTRY signals and closed by EXIT signals.

positions = strategy.get_positions()
for p in positions:
    print(f"{p.symbol}: qty={p.quantity}  unrealized_pnl={p.unrealized_pnl:.2f}")

Trade

Each submitted signal creates a trade record in the platform database (trade_strategy_trade). Trade records track:

• Entry price and timestamp
• Exit price and timestamp (when closed)
• Status: PENDING → APPROVED → CLOSED or REJECTED
• Realized P&L (set when the position is closed)

---

Signal Lifecycle

ENTRY

An ENTRY signal opens a new position. The platform creates a trade record with direction=ENTRY and status=APPROVED.

Signal(
    symbol="AAPL",
    asset_class="equity",
    quantity=100,
    side="BUY",
    signal_type="ENTRY",
    metadata={"reason": "rsi_oversold", "rsi": 27.4},
)

The platform deduplicates ENTRY signals — if an open ENTRY already exists for the same strategy + symbol + side + date, the duplicate is silently skipped.

EXIT

An EXIT signal closes an existing position. The platform:

1. Creates an EXIT trade record with direction=EXIT
2. Finds all open ENTRY trades for strategy + symbol + side
3. Marks them status=CLOSED and links them via exit_trade_id
4. Sets exit_price from the latest market data

Always size exits from live position data — never hardcode quantity:

positions = self.get_positions()
pos_map = {p.symbol: p for p in positions}

for symbol, pos in pos_map.items():
    if should_exit(symbol):
        signals.append(Signal(
            symbol=symbol,
            asset_class="equity",
            quantity=abs(pos.quantity),          # always positive
            side="SELL" if pos.quantity > 0 else "BUY",
            signal_type="EXIT",
            metadata={"reason": "momentum_reversal"},
        ))

EXIT signals are not deduplicated — each EXIT creates a new record and closes matching open entries.

ADJUST

An ADJUST signal modifies the size of an existing position without fully closing it. Use for rebalancing.

---

Realized vs Unrealized P&L

Unrealized P&L is the mark-to-market gain or loss on open positions:

unrealized_pnl = (current_price - avg_entry_price) × quantity

Realized P&L is locked in when a position is closed via an EXIT signal:

realized_pnl = (exit_price - entry_price) × quantity - commission - fees

Fetch both from the platform:

from aptis_strategy_sdk import StrategyAnalytics

analytics = StrategyAnalytics(client, "Theme", "My Strategy")
pnl = analytics.realized_vs_unrealized()
print(f"Realized:   ${pnl['realized_pnl']:+,.2f}")
print(f"Unrealized: ${pnl['unrealized_pnl']:+,.2f}")
print(f"Total:      ${pnl['total_pnl']:+,.2f}")

---

Metadata: Dashboards and AI

The metadata field is a free-form dict stored with every signal. It is never used for execution — it powers dashboards, AI explanations, and attribution analytics.

Recommended structure

from aptis_strategy_sdk import RegimeMetadata, FactorScoreMetadata, RiskSnapshotMetadata

metadata = {
    # Plain-text rationale — shown in AI explanation panels (max 300 chars)
    "rationale": "RSI 27.4 in confirmed bull regime; R/R 2.1:1 at current vol.",

    # Regime context — drives regime overlay on P&L charts
    "regime": RegimeMetadata(
        label="BULL",           # BULL | BEAR | STRESS | CHOP
        confidence=0.81,
        indicators={"adx": 29.3, "vix": 15.1},
    ).to_dict(),

    # Factor attribution — drives bar charts in dashboard
    "factor_scores": FactorScoreMetadata(
        momentum=0.78, quality=0.62, volatility=0.41,
    ).to_dict(),

    # Risk snapshot at signal time
    "risk_metrics": RiskSnapshotMetadata(
        portfolio_var_1d=0.0074,
        max_drawdown=-0.038,
        sharpe_trailing=1.82,
    ).to_dict(),

    # Execution context — for TCA and post-trade analytics
    "execution_context": {
        "trigger": "daily_close",
        "bar_timeframe": "1D",
        "spread_bps": 3.8,
    },

    # Model provenance — for audit trail
    "model_version": "my-strategy-v2.1",

    # Feature snapshot — for AI explainability (key features only)
    "feature_snapshot": {
        "rsi_14": 27.4,
        "momentum_20d": 0.034,
        "atr_14": 2.81,
    },

    # Free-form annotations
    "annotations": ["earnings-clear", "high-conviction"],
}

Frontend conventions

Key	Dashboard use
rationale	AI explanation panel — plain text
regime.label	Badge colour: BULL=green, BEAR=red, STRESS=orange, CHOP=grey
regime.confidence	Confidence gauge widget
factor_scores	Horizontal bar chart — values in [0, 1]
risk_metrics.portfolio_var_1d	Risk gauge — format as 0.74%
risk_metrics.sharpe_trailing	Sharpe badge on strategy card
execution_context.spread_bps	TCA table
model_version	Audit trail tooltip
feature_snapshot	Expandable feature table in signal detail drawer
annotations	Tag chips on signal row
tags (Signal field)	Filter chips in signal feed — use key:value format

Rules

1. All values must be JSON-serialisable (str, int, float, bool, list, dict — no datetime, no numpy types).
2. Use the typed helpers (RegimeMetadata, FactorScoreMetadata, RiskSnapshotMetadata) — they call .to_dict() and strip None fields automatically.
3. Keep rationale under 300 characters.
4. feature_snapshot should contain only the features that directly drove the signal.

---

Institutional Execution Fields

For institutional clients, signals support additional execution controls:

from aptis_strategy_sdk import Signal, Urgency, ExecutionStyle

signal = Signal(
    symbol="BTC/USD",
    asset_class="crypto",
    quantity=0.5,
    side="BUY",
    signal_type="ENTRY",
    # Sizing
    notional_usd=18_500.0,
    portfolio_weight=0.06,
    # Execution quality
    confidence=0.82,
    urgency=Urgency.HIGH,
    execution_style=ExecutionStyle.TWAP,
    time_in_force="DAY",
    max_slippage_bps=15.0,
    limit_price=37_200.0,
    # Routing
    broker_route="PRIME",
    reduce_only=False,
    tags=["momentum", "regime:bull"],
)

All institutional fields are optional. Unset fields are omitted from the JSON payload — no null noise.

Field	Type	Constraint	Description
notional_usd	float	—	Dollar value of the order
portfolio_weight	float	—	Fraction of NAV
confidence	float	[0, 1]	Signal confidence score
urgency	Urgency	—	LOW NORMAL HIGH
time_in_force	str	—	DAY GTC IOC FOK
max_slippage_bps	float	≥ 0	Max acceptable slippage
limit_price	float	—	Limit price
stop_price	float	—	Stop price
execution_style	ExecutionStyle	—	MANUAL MARKET LIMIT VWAP TWAP POV
broker_route	str	—	Routing destination
reduce_only	bool	—	Only reduce existing position
tags	list[str]	—	Filter labels

---

Execution Feedback

Use the execution feedback loop to measure and improve strategy performance:

from aptis_strategy_sdk import StrategyAnalytics
from datetime import date, timedelta

analytics = StrategyAnalytics(client, "Theme", "My Strategy")
start = date.today() - timedelta(days=30)

# Full diagnostics snapshot
diag = analytics.diagnostics(start, date.today())
print(f"Fill rate:       {diag.fill_rate:.1%}")
print(f"Win rate:        {diag.win_rate:.1%}")
print(f"Signal hit rate: {diag.signal_hit_rate:.1%}")
print(f"Total P&L:       ${diag.total_pnl:+,.2f}")
print(f"Gross exposure:  ${diag.gross_exposure_usd:,.2f}")

# Individual fills
fills = client.get_trade_fills("Theme", "My Strategy", start)
for f in fills[:5]:
    print(f"{f.trade_date}  {f.side} {f.quantity} {f.symbol} @ {f.fill_price:.4f}")

# Aggregated execution report
report = client.get_execution_reports("Theme", "My Strategy", start)
print(f"Fill rate: {report.fill_rate:.1%}  Rejections: {report.total_rejections}")

# Per-signal outcomes
results = client.get_signal_results("Theme", "My Strategy", start)
from aptis_strategy_sdk import summarise_signal_results
summary = summarise_signal_results(results)
print(summary)

---

Strategy Lifecycle Hooks

Override these methods to add monitoring, alerting, and pre-flight checks:

class MyStrategy(Strategy):

    def before_run(self, run_date):
        """Called before generate_signals. Use for pre-flight checks."""
        config = self.get_config()
        self.logger.info("NAV=%.0f  enabled=%s", config["nav"], config["enabled"])

    def after_run(self, run_date, results):
        """Called after all signals are submitted."""
        self.logger.info(
            "submitted=%d  skipped=%d  dry_run=%s",
            results["submitted"], results["skipped"], results["dry_run"],
        )

    def on_error(self, run_date, exc):
        """Called on unhandled exception. Exception is always re-raised."""
        self.logger.critical("Strategy error on %s: %s", run_date, exc, exc_info=True)
        # Add PagerDuty / SNS / Slack alerting here

---

Configuration

Environment variables

Variable	Required	Default	Description
APTIS_API_KEY	Yes	—	API key
APTIS_API_URL	No	http://localhost:8082	Backend URL
APTIS_ACCOUNT	No	—	Account name
APTIS_TIMEOUT	No	30	Request timeout (seconds)
APTIS_MAX_RETRIES	No	3	Retry attempts for 429/5xx

Programmatic configuration

from aptis_strategy_sdk import AptisConfig, AptisClient

# From environment (recommended)
client = AptisClient.from_env()

# Explicit config
config = AptisConfig(
    api_key="your_key",
    base_url="https://your-platform-url",
    account="Theme",
    timeout=60,
    max_retries=5,
    backoff_factor=1.0,
)
client = AptisClient(_config=config)

HTTP resilience

The client automatically retries on 429, 500, 502, 503, 504 and connection errors using exponential backoff with jitter. Configure via max_retries and backoff_factor.

---

API Reference

AptisClient

Method	Description
from_env()	Construct from environment variables
submit_signal(signal, strategy_name, account)	Submit a trading signal
get_config(strategy_name)	Get NAV, funds, portfolio_weight, enabled
get_universe(strategy_name)	Get registered symbol universe
get_quotes(symbols, asset_class)	Get market quotes
get_bars(symbol, asset_class, timeframe, start, end)	Get OHLCV bars
get_features(symbols, date, features)	Get Dagster-calculated features
get_positions(account, strategy_name)	Get current positions
get_pnl(account, start_date, end_date, strategy_name)	Get P&L summary
get_trade_fills(account, strategy_name, start_date, end_date)	Get individual fills
get_execution_reports(account, strategy_name, start_date, end_date)	Get aggregated execution report
get_signal_results(account, strategy_name, start_date, end_date)	Get per-signal outcomes
register_strategy(strategy_name, nav, funds, ...)	Register or update strategy config
set_universe(strategy_name, symbols, asset_class)	Set symbol universe
reset_strategy(strategy_name)	Reset strategy: close all trades, clear signals
set_optimizer_overrides(strategy_name, overrides)	Set optimizer parameter overrides
get_optimizer_overrides(strategy_name)	Get current optimizer overrides
update_optimizer_config(strategy_name, **fields)	Update NAV, funds, weight, enabled

Strategy base class

Method	Description
generate_signals(run_date)	Override — return List[Signal]
run(run_date)	Execute full lifecycle, return result dict
before_run(run_date)	Hook — called before generate_signals
after_run(run_date, results)	Hook — called after submission
on_error(run_date, exc)	Hook — called on unhandled exception
get_config()	Fetch strategy config from platform
is_enabled()	Return True if strategy is enabled
get_account()	Return account name
submit_signals(signals)	Validate and submit a list of signals
get_features(symbols, run_date, features)	Fetch Dagster features
get_positions()	Get positions for this strategy
get_pnl(start_date, end_date)	Get P&L for this strategy

StrategyAnalytics

Method	Description
diagnostics(start_date, end_date)	Full StrategyDiagnostics snapshot
fill_rate(start_date, end_date)	Fraction of signals that filled
win_rate(start_date, end_date)	Fraction of closed trades with pnl > 0
signal_hit_rate(start_date, end_date)	Fraction of ENTRY signals that became profitable
average_slippage_bps(start_date, end_date)	Mean slippage in basis points
realized_vs_unrealized()	Realized and unrealized P&L dict
exposure()	Long/short/gross/net exposure in USD

---

Integration Patterns

Scheduled daily strategy

# plugins/my_strategy.py
import os
from datetime import date
from aptis_strategy_sdk import AptisClient, Strategy, Signal
from aptis_strategy_sdk.exceptions import AptisError

class MyStrategy(Strategy):
    def generate_signals(self, run_date: date):
        ...

if __name__ == "__main__":
    client = AptisClient.from_env()
    strategy = MyStrategy(
        "My Strategy", client,
        account=os.getenv("APTIS_ACCOUNT", "Theme"),
    )
    try:
        result = strategy.run(date.today())
    except AptisError as e:
        # Log and alert — do not swallow
        raise

Dry-run in CI

DRY_RUN=true APTIS_API_KEY=ci_key python plugins/my_strategy.py

dry_run = os.getenv("DRY_RUN", "false").lower() == "true"
strategy = MyStrategy("My Strategy", client, "Theme", dry_run=dry_run)

Multi-symbol universe from platform

def generate_signals(self, run_date):
    # Always fetch universe from platform — never hardcode
    universe = self.client.get_universe(self.name) or FALLBACK_UNIVERSE
    features = self.get_features(universe, run_date, ["momentum_20d"])
    ...

Sizing from NAV

def generate_signals(self, run_date):
    config = self.get_config()
    nav = config.get("nav", 1_000_000)
    weight = config.get("portfolio_weight", 0.05)
    notional = nav * weight
    quantity = round(notional / current_price, 4)
    ...

---

Best Practices

1. Always use AptisClient.from_env() — never hardcode API keys or URLs.
2. Check is_enabled() or let run() do it — run() skips automatically when disabled.
3. Size exits from live positions — call self.get_positions(), never hardcode quantity.
4. Use dry_run=True in CI and staging — validates signals without submitting.
5. Override on_error — add alerting (PagerDuty, SNS, Slack) for production strategies.
6. Populate metadata for every signal — enables AI explanations and dashboard attribution.
7. Use tags for structured filtering — format as key:value (e.g. "regime:bull").
8. Declare days_required in your scheduler registration — Dagster uses it to determine history depth.
9. Catch AptisError at the entry point — all SDK exceptions inherit from it.
10. Run StrategyAnalytics.diagnostics() weekly — close the feedback loop.

---

Error Handling

from aptis_strategy_sdk.exceptions import (
    AptisError,           # base — catches everything
    AuthenticationError,  # 401 / 403 — bad key or account
    ConfigurationError,   # missing env vars or invalid config
    ValidationError,      # signal field constraint violated
    APIError,             # non-retryable 4xx
    RateLimitError,       # 429 after all retries
    ServerError,          # 5xx after all retries
)

try:
    result = strategy.run(date.today())
except AuthenticationError:
    # Check APTIS_API_KEY and APTIS_ACCOUNT
    raise
except ValidationError as e:
    # Fix the signal field described in str(e)
    raise
except ServerError as e:
    # Backend is down — alert and retry later
    print(f"HTTP {e.status_code}: {e}")
    raise
except AptisError:
    # Catch-all for any other SDK error
    raise

---

Troubleshooting

Symptom	Cause	Fix
ConfigurationError: APTIS_API_KEY	Env var not set	export APTIS_API_KEY=your_key
AuthenticationError (401)	Wrong API key	Verify key with Richard Chung
AuthenticationError (403)	Key not authorised for account	Check APTIS_ACCOUNT matches key
ValidationError: quantity must be > 0	Negative or zero quantity	Use abs(position.quantity) for exits
ServerError on signal submit	Ticker missing from market_data_ticker	INSERT INTO market_data_ticker (ticker, asset_class) VALUES (...)
Strategy not in frontend dropdown	Not registered in plugin_strategies	Restart scheduler — register_all_plugins() runs on startup
No trades generated	Insufficient market data	Backfill: python cli.py load-twelvedata-daily TICKER --days N
fill_rate is 0	All signals rejected	Check get_signal_results() for rejection_reason
Stale strategy tree in UI	5-minute cache on endpoint	Wait 5 min or restart backend

---

Examples

File	Description
examples/rsi_strategy.py	Simple RSI mean-reversion strategy
examples/momentum_strategy.py	Full production strategy with all best practices
examples/best_practices_strategy.py	Lifecycle hooks, dry-run, structured metadata
examples/institutional_signal.py	All three signal patterns: basic, execution-aware, analytics-rich
examples/execution_feedback.py	Fill rate, win rate, diagnostics, exposure

---

Support

Contact: Richard Chung