Algorithmic Trading System

def scannerData(self, reqId: int, rank: int, contractDetails: ContractDetails, ... ):
    contract = contractDetails.contract
    
    scan_appearance = ScanAppearance(
        scan_code=self.scanner_id_to_code[reqId],
        time_added=datetime.now(),
        time_exited=None            
    )

    with self.lock:
        if contract.symbol not in self.current_scan:
            self.current_scan[contract.symbol] = ScannedStock(
                contract=contract,
                scan_appearances=[scan_appearance]
            )
        else:
            self.current_scan[contract.symbol].scan_appearances.append(scan_appearance)
            
def scannerDataEnd(self, reqId: int):
    print("ScannerDataEnd for", self.scanner_id_to_code[reqId])
    # Current scan only adds things that are not in our watchlist
    print("current scan", self.current_scan.keys())
    
    # Wake up main thread to merge callback data
    with self.lock:
        self.scans_pending -= 1
        
        print("scan finished", self.scanner_id_to_code[reqId])
        if self.scans_pending == 0:
            print("all scans done setting event")
            self.scanner_done_event.set()
            self.scans_pending = len(self.my_scan_codes)
            self.scan_to_process = self.current_scan.copy()
            self.current_scan.clear()

def _advance_bar(self, state: LiveBarSub, pb_bar: msg.Bar, historical_call=False) -> None:
    with self.lock:
        if historical_call:
            self.bar_cache.setdefault(state.ticker, []).append(pb_bar)
            state.last_bar = pb_bar
            return

        if state.seeded and not historical_call:
            previous = state.last_bar
            if previous is not None and previous.timestamp != pb_bar.timestamp:
                self.bar_cache.setdefault(state.ticker, []).append(previous)
            state.last_bar = pb_bar
            return

        # using pending bars, update fired out of order
        if not state.seeded and not historical_call:
            previous = state.pending_last_bar
            if previous is not None and previous.timestamp != pb_bar.timestamp:
                state.pending_cache.append(previous)
            state.pending_last_bar = pb_bar
            return
        
def handle_subscribe(self, sender, command_id, sub_req : msg.SubscriptionReq):
    with self.lock:
        state = self.live_bar_subs.get(sub_req.ticker)

        if state is not None:
            state.subscribers.add(sender)
            joined_mid_seed = not state.seeded # joined during a historical backfill
            if joined_mid_seed:
                state.pending_senders.append((sender, command_id))
            else:
                self._send_cached_bars(sender, command_id, sub_req.ticker)

        else:
            # first subscriber
            req_id = self._next_req_id()
            state = LiveBarSub(req_id=req_id, ticker=sub_req.ticker, subscribers={sender})
            state.seeded = False
            self.live_bar_subs[sub_req.ticker] = state
            self.live_by_reqid[req_id] = state
            self.bar_cache[sub_req.ticker] = []

            self.seeding_acks[state.req_id] = (sender, command_id)
            contract = Contract()
            contract.symbol = sub_req.ticker 
            contract.secType = "STK"
            contract.exchange = "SMART"
            contract.currency = "USD"
            self.reqHistoricalData(state.req_id, contract, "", self._duration_since_cache_start(), "1 min", "TRADES", 0, 2, True, [])

def historicalData(self, reqId: int, bar: BarData):
    state = self.live_by_reqid.get(reqId)
    if state is None:
        return
    pb_bar = self._to_pb_bar(state.ticker, bar)
    self._advance_bar(state, pb_bar, historical_call=True)

def historicalDataEnd(self, reqId: int, start: str, end: str):
    state = self.live_by_reqid.get(reqId)
    if state is None:
        logger.error("HistoricalDataEnd ran for a ticker state that does not exist")
        return

    logger.info(f"Number of historical bars for {state.ticker} : {len(self.bar_cache[state.ticker])}")

    with self.lock:
        state.seeded = True
        if state.pending_cache:
            logger.warning(f"historicalDataUpdate fired before historicalDataEnd... reconciling live cache of length f{len(state.pending_cache)}")
            # merge caches TODO check no duplicates in live cache
            self.bar_cache[state.ticker] += state.pending_cache

            # Update the last bar with the one from live
            if state.pending_last_bar is None: 
                logger.error(f"{state.ticker} last pending bar is None, but pending cache is populated") 
            elif state.last_bar is None:
                logger.warning(f"{state.ticker} No historical data recevied")
            elif not (self._bar_ts_key(state.pending_last_bar) >= self._bar_ts_key(state.last_bar)):
                logger.error("live bar cache (fired out of order) is older than historical update")

            state.last_bar = state.pending_last_bar

    with self.lock:
        waiters = state.pending_senders
        state.pending_senders = []

    # Cache is now fully built -- send everyone waiting on it, in the order
    # they arrived: the original first subscriber, then each mid-seed joiner.
    logger.info(f"{state.ticker} Cache fully built")
    
    entry = self.seeding_acks.pop(reqId, None)
    if entry is not None:
        sender, command_id = entry
        self._send_cached_bars(sender, command_id, state.ticker)
    for sender, command_id in waiters:
        self._send_cached_bars(sender, command_id, state.ticker)

def historicalDataUpdate(self, reqId: int, bar: BarData):
    state = self.live_by_reqid.get(reqId)
    if state is None:
        logger.error("Historical Update recevied for a state that has not been created")
        return
    pb_bar = self._to_pb_bar(state.ticker, bar)

    self._advance_bar(state, pb_bar)
    self.pub_socket.send_multipart([f"BAR:{state.ticker}".encode(), pb_bar.SerializeToString()])

message Bar {
  string ticker = 1;
  google.protobuf.Timestamp timestamp = 2;
  float open = 3;
  float close = 4;
  float high = 5;
  float low = 6;
  float volume = 7;
  int32 num_trades = 8;
}

message BarBatch {
  repeated Bar bars = 1;
  bool is_end = 2;
  string error = 3;
}

enum SubscriptionAction {
  SUBSCRIBE = 0;
  CANCEL = 1;
}

enum DataType {
  BAR = 0;
  TICK = 1;
}

message SubscriptionReq {
  string ticker = 1;
  SubscriptionAction action = 2;
  DataType data_type = 3;
}

class LiveMarketDataClient:
    def __init__(self, strategy_id: str, logger: logging.Logger, pn: PushNotification, host=IB_HOST):
        self.strategy_id = strategy_id
        self.logger = logger
        self.pn = pn

        command_addr = f"tcp://{host}:{MARKET_DATA_SUBSCRIPTION_PORT}"
        pub_addr = f"tcp://{host}:{MARKET_DATA_BAR_PORT}"

        self.book = MarketDataBook()
        self._network = MarketDataNetworkThread(command_addr, pub_addr, self.book, logger, pn, strategy_id)
        self._network.start()

    def subscribe_bars(self, ticker: str, timeout_s: float = SUBSCRIBE_TIMEOUT_S) -> bool:
        """
        Subscribes to the live topic FIRST so live bars start accumulating
        in the book immediately, then requests backfill and merges it in.
        This ordering means there's no join-race window to lose a bar in --
        whatever backfill doesn't cover, the live feed already caught.
        Blocks until backfill fully arrives. Returns False on timeout or
        an error from the master.
        """
        topic = f"BAR:{ticker}".encode()
        self._network.subscribe_topic(topic)

        req = msg.SubscriptionReq(ticker=ticker, action=msg.SUBSCRIBE, data_type=msg.BAR)
        command_id, q = self._network.send_and_collect(req.SerializeToString())

        bars: List[Bar] = []
        while True:
            try:
                payload = q.get(timeout=timeout_s)
            except queue.Empty:
                self._network.drop_pending(command_id)
                self._network.unsubscribe_topic(topic)
                self.logger.error(f"Timed out waiting for backfill: {ticker}")
                self.pn.send_notif(f"@everyone MARKET DATA CLIENT TIMEOUT: backfill {ticker}")
                return False

            batch = msg.BarBatch()
            batch.ParseFromString(payload)
            if batch.error:
                self._network.drop_pending(command_id)
                self._network.unsubscribe_topic(topic)
                self.logger.error(f"Backfill failed for {ticker}: {batch.error}")
                return False

            bars.extend(_to_domain_bar(pb) for pb in batch.bars)
            if batch.is_end:
                self._network.drop_pending(command_id)
                break

        self.book.merge_backfill(ticker, bars)
        return True

# ---- thread body -- all zmq I/O happens here, nowhere else ----
def _run(self):    
    try:
        while not self._stop_event.is_set():
            socks = dict(poller.poll(POLL_INTERVAL_MS))
            if cmd_sock in socks:
                self._handle_command_reply(cmd_sock)
            if sub_sock in socks:
                self._handle_data(sub_sock)
            self._drain_outgoing(cmd_sock)
            self._drain_topic_actions(sub_sock)
    finally:
        cmd_sock.close()
        sub_sock.close()
        context.term()

def _drain_outgoing(self, cmd_sock: zmq.Socket):
    while True:
        try:
            cmd = self._outgoing.get_nowait()
        except queue.Empty:
            return
        cmd_sock.send_multipart([b"", cmd.command_id.encode(), cmd.payload])

def _drain_topic_actions(self, sub_sock: zmq.Socket):
    while True:
        try:
            action, topic = self._topic_actions.get_nowait()
        except queue.Empty:
            return
        opt = zmq.SUBSCRIBE if action == "SUB" else zmq.UNSUBSCRIBE
        sub_sock.setsockopt(opt, topic)

class MarketDataBook:
    def __init__(self):
        self._lock = threading.Lock()
        self._history: Dict[str, Deque[Bar]] = {}
        self._current: Dict[str, Bar] = {}
        self._indicators: Dict[str, List[Indicator]] = {}

    # ---- writers ----

    def apply_live_bar(self, ticker: str, bar: Bar) -> None:
        with self._lock:
            current = self._current.get(ticker)
            completed = None
            if current is not None and current.timestamp != bar.timestamp:
                completed = current
                self._history.setdefault(ticker, deque()).append(completed)
            self._current[ticker] = bar
            indicators = list(self._indicators.get(ticker, ()))

        if completed is not None:
            for indicator in indicators:
                indicator.update(completed)

    def merge_backfill(self, ticker: str, backfill_bars: List[Bar]) -> None:
        """Called once backfill arrives, whenever that is relative to the
        live feed. Only inserts backfill bars for periods not already
        observed live; never regresses `current` if live is further along."""
        if not backfill_bars:
            return
        with self._lock:
            history = self._history.setdefault(ticker, deque(maxlen=DEFAULT_HISTORY_MAXLEN))
            current = self._current.get(ticker)
            known = {b.timestamp for b in history} | ({current.timestamp} if current else set())

            *closed, last_bar = backfill_bars
            missing = [b for b in closed if b.timestamp not in known]

            if missing:
                merged = sorted(list(history) + missing, key=lambda b: b.timestamp)
                maxlen = history.maxlen
                history.clear()
                history.extend(merged[-maxlen:] if maxlen else merged)

            if current is None or last_bar.timestamp >= current.timestamp:
                self._current[ticker] = last_bar

...
while True:
    now = datetime.now().time()
    if now > time(20, 0): # 8:00 PM
        break

    try:
        sender, empty,  command_id, order_binary = order_socket.recv_multipart()
        order_msg = msg.TradeOrder()
        order_msg.ParseFromString(order_binary)

        logger.info(f"Message Received {order_msg.action} {order_msg.ticker}")

        # Send a single order
        if order_msg.classification == msg.OrderAction.NEW:
            client.send_order(order_msg,sender, command_id)

        elif order_msg.classification == msg.OrderAction.DELETE:
            client.cancel_existing_order(order_msg.order_id, sender, command_id)
            
        elif order_msg.classification == msg.OrderAction.MODIFY:
            client.send_order(order_msg,sender, command_id, modify=True)
        
        elif order_msg.classification == msg.OrderAction.GETID:
            start_id, end_id = client.get_order_id_slice(int(order_msg.qty))
            resp = msg.Ticket(
                order_id_start=start_id,
                order_id_end = end_id
            )
            order_socket.send_multipart([sender, b"", resp.SerializeToString()])

    except zmq.Again:
        # It just loops back up to check 'is_market_open' again
        continue

enum OrderAction {
    NEW = 0;
    MODIFY = 1;
    GETID = 2;
    DELETE = 3;
}

message TradeOrder {
  OrderAction classification = 1;
  string ticker = 2;
  string action = 3;
  float qty = 4;
  string order_type = 5;
  // Stop/StopLimit/Limit Orders
  optional double auxprice = 6;
  optional double lmt_price = 7;
  // To cancel or modify an order
  optional int32 order_id = 8;
  // Trailing Stop Orders
  optional float trailingPercent = 9;
  optional int32 trailStopPrice = 10;

}

message TradeUpdate {
    // Unique identifiers
    int32 order_id = 1;
    string ticker = 2;
    // Status info
    string status = 3;
    // Order Info
    string order_type = 4;
    string action = 5;  
    // Quantity tracking
    double total_qty = 6;
    double filled_qty = 7;
    double remaining_qty = 8;   
    // Pricing
    double avg_fill_price = 9;
    double last_fill_price = 10; // Price of the most recent execution
    // Updates for limit and stop orders
    optional double auxprice = 11;
    optional double lmt_price = 12;   
    // Metadata
    string timestamp = 13;      // ISO format or Unix
    optional string error_message = 14;
}

def send_command(self, payload: bytes, parse: Callable[[bytes], object],
                    context_msg: str, timeout_s: float = ACK_TIMEOUT_S):
    """
    Enqueue a command for the network thread to send, then block the *calling* thread
    (not the network thread) until its ack arrives or timeout_s elapses.
    Returns None on timeout, after logging and alerting -- never raises.
    """
    future: concurrent.futures.Future = concurrent.futures.Future()
    command_id = uuid.uuid4().hex

    with self._pending_lock:
        self._pending[command_id] = _PendingCommand(future=future, parse=parse, 
                                                    context_msg=context_msg)

    self._outgoing.put(_OutgoingCommand(command_id=command_id, payload=payload))

    try:
        return future.result(timeout=timeout_s)
    except concurrent.futures.TimeoutError:
        with self._pending_lock:
            self._pending.pop(command_id, None)
        self._logger.error()
        self._pn.send_notif()
        return None

class OrderBook:
    """
    Locked dict of order_id -> OrderRecord, plus a derived net-position view.

    Written to only by the network thread (via `apply_update`).
    Read from by strategy threads (via `get_status` / `wait_for_terminal` /
    `get_position` / `get_all_positions` / `get_open_order_ids`).
    """

    def __init__(self):
        self._lock = threading.Lock()
        self._orders: Dict[int, OrderRecord] = {}
        # Net open position by ticker, derived from fills seen via apply_update.
        # Positive = long, negative = short. Absent key == flat.
        self._positions: Dict[str, float] = {}

    def apply_update(self, update: msg.TradeUpdate) -> None:
        with self._lock:
            record = self._orders.setdefault(update.order_id, OrderRecord(order_id=update.order_id))
            if _is_duplicate(record.latest, update):
                return

class LiveOrderClient():
    def __init__(self, strategy_id: str, logger: logging.Logger, pn: PushNotification, host=IB_HOST):
        self.strategy_id = strategy_id
        self.logger = logger

        server_addr = f"tcp://{host}:{ORDER_SERVICE_PORT}"
        self.order_book = OrderBook()
        self._network = OrderNetworkThread(server_addr, self.order_book, logger, pn, strategy_id)
        self._network.start()

    # ---- commands: fast acks only, never block on a fill ----

    def placeOrder(self, order: msg.TradeOrder) -> Optional[msg.TradeUpdate]:

        resp = self._network.send_command(
                    payload=order.SerializeToString(),
                    parse=_parse_trade_update,
                    context_msg=f"placeOrder {order.ticker} {order.action} {order.qty}",
                )
        
        return resp

logger = logging.getLogger(__name__)

# Create scheduler
scheduler = BlockingScheduler(
    executors={
        'default': ThreadPoolExecutor(max_workers=10),   # I/O-bound jobs (API calls, DB writes)
        'processpool': ProcessPoolExecutor(max_workers=4),  # reserved for future CPU-heavy jobs
    },
    job_defaults={
        'coalesce': True,
        'max_instances': 1,
    }
)

### Update Universe ###
scheduler.add_job(
    runner,
    'interval',
    weeks=2,
    start_date='2026-08-27 11:30:00',
    kwargs={"_func": get_all_listed,
            "_job_name": "Request Target Universe Float"},
    id='get_all_listed_stocks',
    executor='default'
)

scheduler.add_job(
    runner, 
    'cron', 
    hour=22, 
    minute=30, 
    kwargs={
        "_func": run_backfill_session, 
        "_job_name": "Fill data",
        "session_minutes": 60
    },
    id='run_daily_bars_backfill',
    executor='default'
)

def on_fill_executed(event):
    if event.job_id == 'run_daily_bar_backfill':
        # Start downstream dependant jobs
        scheduler.add_job(
            runner, 
            kwargs={
                "_func": generateGapWatchlist, 
                "_job_name": "Generate Gap Watchlist"
            },
            executor='default', 
            id='gap_watchlist_run')
    

def on_fill_error(event):
    if event.job_id == 'run_daily_bar_backfill':
        logger.error(f"Fill job failed, skipping all downstream watchlist jobs: {event.exception}")

# Graceful shutdown
def shutdown(signum, frame):
    logger.info("Shutting down...")
    scheduler.shutdown()
    sys.exit(0)

signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)

if __name__ == "__main__":
    configure_logging(service_name="stocks-data-storage")

    logger.info("Starting BlockingScheduler...")
    scheduler.start()

# IB pacing rules, for daily bars
_rates = [
    Rate(25, Duration.SECOND)
]
limiter = Limiter(_rates)

while i < len(pending):
    if datetime.now() >= session_deadline:
        logger.info(f"Time budget reached. {total_done} done this run, {len(pending) - total_done} remain.")
        return

    batch = pending[i:i + BATCH_SIZE]
    deltas = [compute_delta_days(last_dates.get(s), today) for s in batch]

    # One acquire per underlying request this batch will make against IB's
    # shared budget -- blocks (sleeps internally) until there's room under
    # both the 10-min window and the 2-sec burst rule.
    for _ in batch:
        limiter.try_acquire("ib_historical_data")

    logger.info(f"Processing Batch : {i//BATCH_SIZE}/{ceil(len(pending)/BATCH_SIZE)}")

    bars_by_symbol = ib_client.get_historical_data(
        tickers=batch,
        deltas=deltas,
        bar_duration="1 day",
    )

    if ib_client.pacing_violation:
        logger.warning(f"Pacing Volation Recevied. Sleeping and reprocessing Batch : {i//BATCH_SIZE}/{ceil(len(pending)/BATCH_SIZE)}")
        continue

    # Resets each call so pull it before you call
    invalid_tickers = ib_client.invalid_tickers

    if invalid_tickers:
        upsert_invalid_tickers(invalid_tickers)

    for symbol, bars in bars_by_symbol.items():
        if not bars:
            logger.info(f"No bars returned for {symbol}, will retry next session")
            continue
        upsert_daily_bars(symbol, bars)
        total_done += 1

    logger.info(f"{total_done}/{len(pending)} symbols done this session")
    i += BATCH_SIZE

logger.info(f"All pending symbols processed this session. Total: {total_done}")

from trading_util.db_util import StocksDBConnection

import os
from contextlib import contextmanager
import threading

from psycopg2 import pool

DB_PORT = 5432
DB_HOST='192.168.2.60'

class StocksDBConnection:
    def __init__(self, minconn: int = 1, maxconn: int = 10):
        self._pool = pool.ThreadedConnectionPool(
            minconn, 
            maxconn,
            host=host
            dbname='stocks',
            user='brian',
            password=password,
            port=DB_PORT
        )
        self._semaphore = threading.Semaphore(maxconn)

    # @classmethod
    # def from_env(cls, env_var: str = "STOCKS_DB_DSN", **kwargs) -> "StocksDBConnection":
    #     dsn = os.environ[env_var]
    #     return cls(dsn=dsn, **kwargs)

    def close(self):
        """Call on service shutdown to release all pooled connections cleanly."""
        self._pool.closeall()

    @contextmanager
    def cursor(self, timeout: float | None = None):
        acquired = self._semaphore.acquire(timeout=timeout)
        if not acquired:
            raise TimeoutError("Timed out waiting for a free DB connection")

        conn = self._pool.getconn()
        try:
            with conn.cursor() as cur:
                yield cur
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            self._pool.putconn(conn)
            self._semaphore.release()


class GapWatchlistDB:
    def __init__(self, conn: StocksDBConnection):
        self._conn = conn

    def get_active_watchlist(self) -> dict:
        """{stock: {"date_added", "last_updated", "entry_rsi", "entry_atr", "entry_close",
                    "latest_rsi", "latest_atr", "latest_close"}}"""
        with self._conn.cursor() as cur:
            cur.execute("""
                SELECT stock, date_added, last_updated, entry_rsi, entry_atr, entry_close,
                       latest_rsi, latest_atr, latest_close
                FROM gap_watchlist
            """)
            rows = cur.fetchall()
            return {
                r[0]: {
                    "date_added": r[1],
                    "last_updated": r[2],
                    "entry_rsi": r[3],
                    "entry_atr": r[4],
                    "entry_close": r[5],
                    "latest_rsi": r[6],
                    "latest_atr": r[7],
                    "latest_close": r[8],
                }
                for r in rows
            }

    ...