"""Firstock broker implementation""" import time from typing import Dict, List, Optional from firstock import firstock from .base_broker import BaseBroker from utils.logger import TradingLogger class FirstockBroker(BaseBroker): """Firstock specific implementation""" def __init__(self, credentials: Dict, client_code: str): super().__init__(credentials, client_code) self.user_id = credentials.get('user_id') self.password = credentials.get('password') self.totp = credentials.get('totp') self.vendor_code = credentials.get('vendor_code') self.api_key = credentials.get('api_key') self.logger = TradingLogger(client_code) def login(self) -> bool: """Login to Firstock""" try: response = firstock.login( userId=self.user_id, password=self.password, TOTP=self.totp, vendorCode=self.vendor_code, apiKey=self.api_key ) if response.get('status') == 'success': self.is_logged_in = True self.logger.success(f"Firstock login successful for user: {self.user_id}") return True else: self.logger.error(f"Firstock login failed: {response}") return False except Exception as e: self.logger.error(f"Firstock login error: {e}") return False def place_order(self, symbol: str, action: str, quantity: int, exchange: str, expiry: str, remarks: str) -> Optional[str]: """Place order on Firstock""" try: if symbol == 'SENSEX': trading_symbol = symbol else: pre = symbol.replace("NIFTY", "") instrument = pre.replace("PE", "").replace("CE", "") option_type = "C" if "CE" in pre else "P" trading_symbol = f"NIFTY{expiry}{option_type}{instrument}" transaction_type = 'B' if action.upper() == 'BUY' else 'S' response = firstock.placeOrder( userId=self.user_id, exchange=exchange, tradingSymbol=trading_symbol, quantity=str(quantity), price="0.0", product="M", transactionType=transaction_type, priceType="MKT", retention="DAY", triggerPrice="0", mkt_protection="10", remarks=remarks ) self.logger.success(f"Firstock order placed: {response}, Symbol={trading_symbol}, Qty={quantity}") return response.get('orderId') except Exception as e: self.logger.error(f"Firstock order failed: {e}") return None def place_profit_order(self, trading_symbol: str, quantity: int, exchange: str, transaction_type: str, remarks: str) -> Optional[str]: """Place profit booking order on Firstock""" try: response = firstock.placeOrder( userId=self.user_id, exchange=exchange, tradingSymbol=trading_symbol, quantity=str(quantity), price="0.0", product="M", transactionType=transaction_type, priceType="MKT", retention="DAY", triggerPrice="0", mkt_protection="10", remarks=remarks ) self.logger.success(f"Firstock profit order placed: {response}, Symbol={trading_symbol}") return response.get('orderId') except Exception as e: self.logger.error(f"Firstock profit order failed: {e}") return None def get_positions(self) -> List[Dict]: """Get current positions from Firstock""" try: response = firstock.positionBook(userId=self.user_id) return response.get('data', []) except Exception as e: self.logger.error(f"Failed to get positions: {e}") return [] def get_orders(self) -> List[Dict]: """Get order book from Firstock""" try: response = firstock.orderBook(userId=self.user_id) return response.get('data', []) except Exception as e: self.logger.error(f"Failed to get orders: {e}") return [] def get_quote(self, exchange: str, trading_symbol: str) -> Dict: """Get market quote from Firstock""" try: response = firstock.getQuote( userId=self.user_id, exchange=exchange, tradingSymbol=trading_symbol ) data = response.get('data', {}) return { 'last_price': float(data.get('lastTradedPrice', 0)), 'change': float(data.get('change', 0)) } except Exception as e: self.logger.error(f"Failed to get quote for {trading_symbol}: {e}") return {'last_price': 0, 'change': 0} def get_realized_pnl(self) -> float: """Get total realized PnL from Firstock""" try: positions = self.get_positions() total_pnl = sum(float(pos.get('RealizedPNL', 0)) for pos in positions) return total_pnl except Exception as e: self.logger.error(f"Failed to get realized PnL: {e}") return 0.0 def close_all_positions(self, max_profit_target: int, max_loss_limit: int, dynamic_profit_enabled: bool) -> bool: """Close all open positions based on limits""" if max_profit_target == 0 and max_loss_limit == 0: return False try: positions = self.get_positions() if all(int(pos.get('netQuantity', 0)) == 0 for pos in positions): total_pnl = self.get_realized_pnl() if total_pnl > max_profit_target: self.logger.info(f"MAX PROFIT reached: {total_pnl} > {max_profit_target}") return True elif total_pnl < -max_loss_limit: self.logger.info(f"MAX LOSS reached: {total_pnl} < -{max_loss_limit}") return True return False if not dynamic_profit_enabled: return False total_pnl = sum(float(pos.get('totalPNL', 0)) for pos in positions) should_close = False marker = "" if total_pnl > max_profit_target: should_close = True marker = "ProfitLimit" elif total_pnl < -max_loss_limit: should_close = True marker = "LossLimit" if should_close: self.logger.info(f"FORCE CLOSE ALL POSITIONS - {marker}") for position in positions: quantity = int(position.get('netQuantity', 0)) if quantity != 0: trading_symbol = position.get('tradingSymbol') exchange = position.get('exchange') transaction_type = 'S' self.place_profit_order( trading_symbol=trading_symbol, quantity=quantity, exchange=exchange, transaction_type=transaction_type, remarks=f"FORCE_CLOSE_{marker}" ) time.sleep(0.5) return True return False except Exception as e: self.logger.error(f"Failed to close positions: {e}") return False