SMA Crossover: Lab Python vs. IBKR Python

# SMA Crossover — Trading Lab Python version
# Paste directly into the Lab code editor (select Python mode)
# strategy_type: sma-crossover

def initialize(context):
    context.symbol      = 'AAPL'
    context.fast_period = 10    # 10-day fast SMA
    context.slow_period = 30    # 30-day slow SMA

def handle_data(context, data):
    # data.history() returns a pandas Series — no API call needed
    prices = data.history(context.symbol, 'price', context.slow_period + 1)
    if len(prices) < context.slow_period + 1:
        return  # Not enough history yet

    # SMA calculations — identical math to the IBKR version below
    fast_now  = prices.iloc[-context.fast_period:].mean()
    slow_now  = prices.iloc[-context.slow_period:].mean()
    fast_prev = prices.iloc[:-1].iloc[-context.fast_period:].mean()
    slow_prev = prices.iloc[:-1].iloc[-context.slow_period:].mean()

    # context.portfolio.positions is a dict — no API call needed
    position = context.portfolio.positions.get(context.symbol)

    # Golden cross: fast crosses ABOVE slow → buy
    if not position and fast_prev <= slow_prev and fast_now > slow_now:
        order_target_percent(context.symbol, 1.0)  # engine handles sizing + fill
        log.info(f"BUY {context.symbol}: golden cross (fast={fast_now:.2f}, slow={slow_now:.2f})")

    # Death cross: fast crosses BELOW slow → sell
    elif position and fast_prev >= slow_prev and fast_now < slow_now:
        order_target_percent(context.symbol, 0.0)  # engine handles sell + fill
        log.info(f"SELL {context.symbol}: death cross (fast={fast_now:.2f}, slow={slow_now:.2f})")

"""
SMA Crossover — IBKR standalone Python version
Uses ib_insync; IB Gateway must be running on port 4002 (paper).

Compare with the Lab version above:
  data.history()                    → get_historical_closes()
  context.portfolio.positions.get() → get_position_size()
  order_target_percent(sym, 1.0)    → place_market_order('BUY', qty)
  log.info()                        → logging.getLogger()
"""

import os
import time
import logging
from datetime import datetime
import pandas as pd
from ib_insync import IB, Stock, MarketOrder, util

# ── Config (read from environment variables for production) ────────
SYMBOL      = os.environ.get('SYMBOL',      'AAPL')
FAST_PERIOD = int(os.environ.get('FAST_PERIOD', '10'))
SLOW_PERIOD = int(os.environ.get('SLOW_PERIOD', '30'))
CAPITAL_PCT = float(os.environ.get('CAPITAL_PCT', '0.95'))
IB_HOST     = os.environ.get('IB_HOST',     '127.0.0.1')
IB_PORT     = int(os.environ.get('IB_PORT', '4002'))   # 4002 = IB Gateway paper
CLIENT_ID   = int(os.environ.get('CLIENT_ID', '1'))

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s %(levelname)s %(message)s')
log = logging.getLogger(__name__)


# ── What the Lab hides: data fetching ─────────────────────────────
# In the Lab: prices = data.history(symbol, 'price', period)
# On IBKR:   you call reqHistoricalData() with explicit parameters

def get_historical_closes(ib: IB, contract: Stock, lookback_days: int) -> pd.Series:
    bars = ib.reqHistoricalData(
        contract,
        endDateTime='',                        # '' = now
        durationStr=f'{lookback_days + 10} D', # +10 buffer for weekends/holidays
        barSizeSetting='1 day',
        whatToShow='TRADES',
        useRTH=True,                           # regular trading hours only
        formatDate=1,
    )
    if not bars:
        raise RuntimeError(f'No data for {contract.symbol}')
    return util.df(bars)['close']  # returns pd.Series — same type as data.history()


# ── What the Lab hides: position tracking ─────────────────────────
# In the Lab: position = context.portfolio.positions.get(symbol)
# On IBKR:   you call ib.positions() and filter by symbol yourself

def get_position_size(ib: IB, symbol: str) -> float:
    for pos in ib.positions():
        if pos.contract.symbol == symbol:
            return pos.position  # positive = long, negative = short
    return 0.0


# ── What the Lab hides: account balance ───────────────────────────
# In the Lab: order_target_percent(sym, 1.0) calculates sizing for you
# On IBKR:   you query cash, compute shares = cash * pct / price yourself

def get_available_cash(ib: IB) -> float:
    summary = {v.tag: v.value for v in ib.accountSummary()}
    return float(summary.get('AvailableFunds', 0))


# ── What the Lab hides: order routing ─────────────────────────────
# In the Lab: order_target_percent(sym, 1.0) sends the order
# On IBKR:   you build a MarketOrder, place it, and poll for fill

def place_market_order(ib: IB, contract: Stock, action: str, qty: int):
    order = MarketOrder(action, qty)
    trade = ib.placeOrder(contract, order)
    for _ in range(10):        # wait up to 10 seconds for fill
        ib.sleep(1)
        if trade.isDone():
            break
    log.info('%s %d %s — status=%s fill=%.2f',
             action, qty, contract.symbol,
             trade.orderStatus.status,
             trade.orderStatus.avgFillPrice or 0)
    return trade


# ── The signal logic — identical to the Lab version ───────────────
# This block maps 1:1 to handle_data() above. The math is the same.

def evaluate_signal(closes: pd.Series, fast: int, slow: int) -> str:
    if len(closes) < slow + 1:
        return 'hold'

    fast_now  = closes.iloc[-fast:].mean()
    slow_now  = closes.iloc[-slow:].mean()
    fast_prev = closes.iloc[:-1].iloc[-fast:].mean()
    slow_prev = closes.iloc[:-1].iloc[-slow:].mean()

    if fast_prev <= slow_prev and fast_now > slow_now:
        return 'buy'   # golden cross
    if fast_prev >= slow_prev and fast_now < slow_now:
        return 'sell'  # death cross
    return 'hold'


# ── Main run — what the platform engine handles automatically ──────

def run():
    ib = IB()
    try:
        ib.connect(IB_HOST, IB_PORT, clientId=CLIENT_ID)
        log.info('Connected to IBKR (host=%s port=%d)', IB_HOST, IB_PORT)

        contract = Stock(SYMBOL, 'SMART', 'USD')
        ib.qualifyContracts(contract)

        closes = get_historical_closes(ib, contract, lookback_days=SLOW_PERIOD + 5)
        log.info('Fetched %d closes for %s. Latest: %.2f', len(closes), SYMBOL, closes.iloc[-1])

        signal = evaluate_signal(closes, FAST_PERIOD, SLOW_PERIOD)
        log.info('Signal: %s', signal)

        held = get_position_size(ib, SYMBOL)
        log.info('Current position: %.0f shares', held)

        if signal == 'buy' and held == 0:
            cash = get_available_cash(ib)
            price = closes.iloc[-1]
            qty = int((cash * CAPITAL_PCT) / price)
            if qty > 0:
                place_market_order(ib, contract, 'BUY', qty)
            else:
                log.warning('Insufficient cash (cash=%.2f price=%.2f)', cash, price)

        elif signal == 'sell' and held > 0:
            place_market_order(ib, contract, 'SELL', int(held))

        else:
            log.info('HOLD — no action (signal=%s held=%.0f)', signal, held)

    except Exception as exc:
        log.error('Fatal error: %s', exc, exc_info=True)
        raise
    finally:
        ib.disconnect()
        log.info('Disconnected')


if __name__ == '__main__':
    run()

# scheduler.py — runs sma_strategy.py every weekday at 4:05 PM ET
import schedule, time, pytz
from datetime import datetime
from sma_strategy import run

ET = pytz.timezone('America/New_York')

def job():
    if datetime.now(ET).weekday() >= 5:   # 5=Sat 6=Sun
        return
    print(f'[{datetime.now(ET).strftime("%H:%M ET")}] Running SMA signal check...')
    run()

schedule.every().day.at('21:05').do(job)  # 4:05 PM ET = 21:05 UTC

print('Scheduler started — waiting for 4:05 PM ET')
while True:
    schedule.run_pending()
    time.sleep(30)