Building Autonomous AI Trading Bots with DeepSeek-V4.1, Tradier & Python

# Architectural Decoupling: Signal vs Execution
"""
[Market Feeds: Tradier / CCXT] 
       │ (WebSockets / REST L2)
       ▼
[Feature Pipeline: Orderflow, VWAP, Volatility, Sentiment]
       │
       ▼
[Reasoning Layer: DeepSeek-V4.1 Signal Engine]
       │ (Outputs Structured TradeHypothesis JSON)
       ▼
[Risk Management Shield (Deterministic Python Gate)]
   ├── Check 1: Max daily account drawdown (< 3.0%)
   ├── Check 2: Fractional Kelly sizing limit (max 2% capital per trade)
   ├── Check 3: Market spread & liquidity filter
       │ (APPROVED)
       ▼
[Order Routing Engine: Tradier / CCXT Broker Gateway]
   ├── Post-only Limit Order
   └── Bracket Order with Trailing Stop & Take-Profit
"""

import os
import requests
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()

TRADIER_TOKEN = os.getenv("TRADIER_SANDBOX_TOKEN")
TRADIER_BASE = "https://sandbox.tradier.com/v1"

@dataclass
class MarketSnapshot:
    symbol: str
    last_price: float
    bid: float
    ask: float
    spread_pct: float
    volume: int

def get_tradier_snapshot(symbol: str) -> MarketSnapshot:
    headers = {
        "Authorization": f"Bearer {TRADIER_TOKEN}",
        "Accept": "application/json"
    }
    resp = requests.get(f"{TRADIER_BASE}/markets/quotes", params={"symbols": symbol}, headers=headers)
    resp.raise_for_status()
    data = resp.json()["quotes"]["quote"]
    
    spread = data["ask"] - data["bid"]
    spread_pct = (spread / data["last"]) * 100
    
    return MarketSnapshot(
        symbol=symbol,
        last_price=float(data["last"]),
        bid=float(data["bid"]),
        ask=float(data["ask"]),
        spread_pct=round(spread_pct, 4),
        volume=int(data["volume"])
    )

# Quick test
# print(get_tradier_snapshot("NVDA"))

import json
from pydantic import BaseModel, Field
from typing import Literal, Optional

class TradeSignal(BaseModel):
    symbol: str
    action: Literal["BUY", "SELL", "HOLD"]
    confidence: float = Field(ge=0.0, le=1.0, description="Confidence score between 0 and 1")
    rationale: str
    suggested_entry: float
    stop_loss_price: float
    take_profit_price: float
    expected_hold_duration_mins: int

SYSTEM_PROMPT = """
You are a Quantitative Risk & Momentum Analyst operating on 5-minute equity data.
Evaluate the provided Technical Feature Vector and Order Flow Imbalance.
Output ONLY valid JSON adhering strictly to the TradeSignal schema.
If confidence is below 0.72 or risk/reward is worse than 1:2.5, action MUST BE 'HOLD'.
"""

def synthesize_trading_signal(client, features: dict) -> TradeSignal:
    prompt = f"Market State: {json.dumps(features, indent=2)}"
    response = client.chat.completions.create(
        model="deepseek-chat", # DeepSeek-V4.1 endpoint
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": prompt}
        ],
        response_format={"type": "json_object"},
        temperature=0.15
    )
    raw_json = json.loads(response.choices[0].message.content)
    return TradeSignal(**raw_json)

class RiskShield:
    def __init__(self, initial_balance: float, max_daily_loss_pct: float = 0.03):
        self.initial_balance = initial_balance
        self.current_balance = initial_balance
        self.max_daily_loss = initial_balance * max_daily_loss_pct
        self.daily_realized_pnl = 0.0
        self.is_circuit_breaker_tripped = False

    def evaluate_order(self, signal: TradeSignal, current_price: float):
        # Rule 1: Circuit breaker check
        if self.is_circuit_breaker_tripped:
            print("🚨 RiskShield: Circuit breaker tripped! All trading suspended.")
            return None
            
        if signal.action == "HOLD":
            return None

        # Rule 2: Risk-to-Reward Ratio Check (minimum 1:2.5)
        potential_loss = abs(current_price - signal.stop_loss_price)
        potential_gain = abs(signal.take_profit_price - current_price)
        if potential_loss <= 0 or (potential_gain / potential_loss) < 2.5:
            print(f"⚠️ RiskShield VETO: R/R ratio insufficient ({potential_gain / (potential_loss or 1):.2f})")
            return None

        # Rule 3: Fractional Kelly Criterion Position Sizing (Half-Kelly)
        win_rate = signal.confidence * 0.65  # Conservative probability calibration
        reward_ratio = potential_gain / potential_loss
        kelly_fraction = win_rate - ((1.0 - win_rate) / reward_ratio)
        safe_kelly = max(0.0, min(kelly_fraction * 0.5, 0.02)) # Cap at 2% max risk per trade

        risk_capital = self.current_balance * safe_kelly
        shares = int(risk_capital / potential_loss)
        if shares < 1:
            print("⚠️ RiskShield VETO: Calculated position size is less than 1 share.")
            return None

        return {
            "symbol": signal.symbol,
            "side": "buy" if signal.action == "BUY" else "sell_short",
            "quantity": shares,
            "price": current_price,
            "stop_loss": signal.stop_loss_price,
            "take_profit": signal.take_profit_price
        }

    def record_pnl(self, pnl: float):
        self.daily_realized_pnl += pnl
        self.current_balance += pnl
        if self.daily_realized_pnl <= -self.max_daily_loss:
            self.is_circuit_breaker_tripped = True
            print("🚨 CRITICAL: Daily max loss breached! Kill-switch engaged.")

import numpy as np
import pandas as pd
import vectorbt as vbt

def run_vectorized_simulation(ohlcv_df: pd.DataFrame):
    fast_ema = vbt.MA.run(ohlcv_df["Close"], window=12, ewm=True)
    slow_ema = vbt.MA.run(ohlcv_df["Close"], window=26, ewm=True)
    rsi = vbt.RSI.run(ohlcv_df["Close"], window=14)

    entries = fast_ema.ma_crossed_above(slow_ema) & (rsi.rsi < 65)
    exits = fast_ema.ma_crossed_below(slow_ema) | (rsi.rsi > 80)

    portfolio = vbt.Portfolio.from_signals(
        ohlcv_df["Close"],
        entries=entries,
        exits=exits,
        init_cash=100000.0,
        fees=0.0005,
        slippage=0.0002,
        freq="5m"
    )

    print(f"Sharpe Ratio:     {portfolio.sharpe_ratio():.2f}")
    print(f"Max Drawdown:     {portfolio.max_drawdown() * 100:.2f}%")
    print(f"Total Return:     {portfolio.total_return() * 100:.2f}%")
    print(f"Win Rate:         {portfolio.trades.win_rate() * 100:.2f}%")
    return portfolio

# Dockerfile for 24/7 Trading Agent
"""
FROM python:3.11-slim

WORKDIR /app
RUN apt-get update && apt-get install -y gcc build-essential curl && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

ENV PYTHONUNBUFFERED=1
CMD ["python", "bot_daemon.py"]
"""

# Telegram Alert Dispatcher Snippet
def send_telegram_alert(token: str, chat_id: str, message: str):
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    payload = {
        "chat_id": chat_id,
        "text": message,
        "parse_mode": "Markdown"
    }
    try:
        requests.post(url, json=payload, timeout=5)
    except Exception as e:
        print(f"Failed to send Telegram alert: {e}")