Issue with Events - Max Drawdown script

[mporenta]

Hey all, I am very new to python and async outside of browser. My goal here is for this to run as a second client " client_id=2", to protect myself from a failure due to tws fat fingers or from client 1 from doing something unintended(lets be real, I am not worried about tws fat fingers).

Either I get infinite "Warning :Loss threshold of -1.00 exceeded with daily PnL: -1776.51 at 2024-08-27 12:48:28" or I get that message once and it quits.
Looking for the Python gods to smile down upon me with pitty. Mocking in the comments accepted and welcome. :)

`import logging
from datetime import datetime
import pytz
from ib_async import IB, MarketOrder

class SimplePnLStrategy:
def init(self, host='127.0.0.1', port=7497, client_id=2, loss_threshold=-1):
self.host = host
self.port = port
self.client_id = client_id

    self.ib = None
    self.pnl = None
    self.positions = {}
    self.loss_threshold = loss_threshold  # Loss threshold for daily PnL

    # Initialize logger
    self.setup_logging()

def setup_logging(self):
    """Setup logging configuration"""
    logging.basicConfig(
        format='%(asctime)s [%(levelname)s] %(message)s',
        level=logging.INFO,
        handlers=[
            logging.FileHandler("trading_strategy.log"),
            logging.StreamHandler()
        ]
    )
    logging.info("Logger initialized.")

def run(self):
    logging.info("Running the strategy...")

    # Establish connection to IB
    self.connect_to_ib()

    # Subscribe to PnL and position updates
    self.request_pnl_updates()
    self.request_position_updates()

    logging.info("Entering main update loop...")

    # Loop indefinitely to monitor PnL and check for loss threshold
    while True:
        self.ib.waitOnUpdate()
        self.ib.sleep(1)

        # Check if loss threshold is reached
        if self.check_loss_threshold():
            self.close_all_positions()

            # Log and exit the loop after closing positions
            logging.info("Loss threshold exceeded, positions closed. Exiting loop.")
            break  # Exit after closing positions



def get_current_time(self):
    """Returns the current timestamp in local time"""
    local_tz = pytz.timezone('America/New_York')  # Adjust for your local timezone
    return datetime.now(local_tz).strftime('%Y-%m-%d %H:%M:%S')

def connect_to_ib(self):
    try:
        self.ib = IB()
        self.ib.connect(self.host, self.port, clientId=self.client_id)
        logging.info("Connected to IB at %s", self.get_current_time())
    except Exception as e:
        logging.error("Failed to connect to IB: %s", str(e))

def request_pnl_updates(self):
    try:
        account = self.ib.managedAccounts()[0]
        self.ib.reqPnL(account)
        self.ib.pnlEvent += self.on_pnl
        logging.info("Subscribed to PnL updates.")
    except Exception as e:
        logging.error("Failed to subscribe to PnL updates: %s", str(e))

def on_pnl(self, pnl):
    """ Store the latest PnL including dailyPnL """
    self.pnl = pnl
    logging.info(
        "PnL updated: Daily PnL: %.2f, Realized PnL: %.2f, Unrealized PnL: %.2f at %s",
        pnl.dailyPnL, pnl.realizedPnL, pnl.unrealizedPnL, self.get_current_time()
    )

def request_position_updates(self):
    try:
        self.ib.reqPositions()
        self.ib.positionEvent += self.on_position
        logging.info("Subscribed to position updates.")
    except Exception as e:
        logging.error("Failed to subscribe to position updates: %s", str(e))

def on_position(self, position):
    """ Store the latest positions """
    symbol = position.contract.symbol
    self.positions[symbol] = position
    logging.info(
        "Position updated: %s - %d units at %s",
        symbol, position.position, self.get_current_time()
    )
    logging.info("Current positions: %s", self.positions)  # Add this line


def check_loss_threshold(self):
    """
    Check if the loss threshold has been reached based on dailyPnL.
    """
    if self.pnl and self.pnl.dailyPnL <= self.loss_threshold:
        logging.warning(
            "Loss threshold of %.2f exceeded with daily PnL: %.2f at %s",
            self.loss_threshold, self.pnl.dailyPnL, self.get_current_time()
        )
        return True
    return False
def close_all_positions(self):
    """ Close all positions when the loss threshold is exceeded """
    logging.info("Attempting to close all positions at %s", self.get_current_time())
    for symbol, position in self.positions.items():
        contract = position.contract
        qty = position.position

        if qty > 0:  # Long position, sell to close
            order = MarketOrder('SELL', qty)
        elif qty < 0:  # Short position, buy to cover
            order = MarketOrder('BUY', abs(qty))
        else:
            logging.info("No position to close for %s at %s", symbol, self.get_current_time())
            continue  # Skip if no position

        # Log the order details before placing it
        logging.info("Placing order for %s to close position of %d units at %s", symbol, qty, self.get_current_time())

        try:
            trade = self.ib.placeOrder(contract, order)
            logging.info("Placed order to close position for %s: %s at %s", symbol, trade, self.get_current_time())
        except Exception as e:
            logging.error("Failed to place order for %s: %s", symbol, str(e))

    logging.info("Completed position closing attempt at %s", self.get_current_time())

if name == "main":
logging.info("Script is starting...")
strategy = SimplePnLStrategy()
strategy.run()
logging.info("Script has finished.")
`

[mporenta]

sorry for the bad markup, I was in a hurry

[jj625]

Better to reference self.ib.positions() directly instead of using self.positions.

ib.positions() is part of "current state", always synchronized with TWS/IBG, so you don't have to do anything to keep it up to date. See notebooks/basics.ipynb for sample usage.

I tested your code using ib.positions(). Works as intended now.

[mporenta]

Thanks! I will try this in the morning.