Metadata-Version: 2.4
Name: aptis-strategy-sdk
Version: 0.5.2
Summary: Python SDK for building plugin trading strategies on the Aptis platform
Author: Richard Chung
Project-URL: Homepage, https://aptis.com
Project-URL: Documentation, https://aptis.com/docs
Classifier: License :: Other/Proprietary License
Classifier: Programming Language :: Python :: 3
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: requests>=2.31.0
Requires-Dist: urllib3>=1.26.0
Requires-Dist: pandas>=2.0.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: black>=23.0.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"
Requires-Dist: pytest-mock>=3.0.0; extra == "dev"

# 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

```bash
pip install aptis-strategy-sdk
```

### 2. Configure

Create a `.env` file in your project directory:

```bash
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

```python
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

```python
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`)

```python
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.

```python
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`.

```python
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:

```python
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:

```python
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

```python
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:

```python
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:

```python
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:

```python
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

```python
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

```python
# 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

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

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

### Multi-symbol universe from platform

```python
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

```python
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

```python
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
