python order manager for trading .

order_manager.py

import json
from datetime import datetime
from typing import List, Dict, Optional
from tabulate import tabulate


class Order:
    def __init__(self, symbol: str, quantity: float, price: float, order_type: str, timestamp: Optional[datetime] = None):
        self.symbol = symbol
        self.quantity = quantity
        self.price = price
        self.order_type = order_type
        self.timestamp = timestamp or datetime.now()
        self.lastUpdateTime = timestamp or datetime.now()
        self.OpenReason ="OpenReason"
        self.CloseReason ="Not Closed , Position Running"
        self.status = "Running"
        self.pnl = 0.0
        self.pnlHigh = 0.0
        self.pnlLow = 0.0
        self.pnlPercentage = 0.0
        self.pnlHighPercentage = 0.0
        self.pnlLowPercentage = 0.0
        self.lastPrice =0.0
        self.UpdateCount =0
        self.PointsMoved =0
        self.quantity_Index =0
        self.setIndex=0
        self.groupID = "Group1"
        self.Comment = "Comment"
        self.Take_profit = 0
        self.Stop_loss = 0
        self.upside_order=False
        self.downside_order=False
        self.estimated_points=0
        self.Update(price)
        self.id= f"{self.symbol}_{self.timestamp.strftime('%Y%m%d_%H%M%S')}"
         # Flag to indicate if the order should be saved to a file

    def Update(self, close_price: float):
        self.lastUpdateTime =  datetime.now()
        self.lastPrice=close_price
       
        self.status = "Running"
        self.UpdateCount +=1
        self.pnl = (close_price - self.price) * self.quantity if self.order_type == "buy" else (self.price - close_price) * self.quantity
        self.pnlPercentage = ((close_price - self.price) / self.price) * 100 if self.order_type == "buy" else ((self.price - close_price) / self.price) * 100
        self.pnl = round(self.pnl,0)
        self.pnlPercentage = round(self.pnlPercentage,6)
        if self.pnl > self.pnlHigh :
            self.pnlHigh=self.pnl
            self.pnlHighPercentage = self.pnlPercentage
        if self.pnl < self.pnlLow :
            self.pnlLow=self.pnl
            self.pnlLowPercentage = self.pnlPercentage
        self.PointsMoved = round(abs(self.lastPrice - self.price),4)
        if self.Take_profit != 0 and self.Stop_loss != 0:
            self.Update_tp_sl()
    
    def Update_tp_sl(self):
        if (self.pnl > self.Take_profit) or (self.pnl < -self.Stop_loss):
             self.close(self.lastPrice)


    def remap(self,value, old_min, old_max, new_min, new_max):
        a=((value - old_min) / (old_max - old_min)) * (new_max - new_min) + new_min
        return round(a,6)

    def get_lifetime(self) -> str:
        lifetime = self.lastUpdateTime - self.timestamp
        hours, remainder = divmod(lifetime.seconds, 3600)
        minutes, seconds = divmod(remainder, 60)
        if hours > 5 :
            desc = " \nRunning Position LifeTime is above 6 hours "
        else :
            desc ="\nRunning Position LifeTime is below 6 Hours "    
        return f"lifetime {hours:02}:{minutes:02}:{seconds:02} ,Position is running since {hours:02} Hours and {minutes:02} minutes."
    
    def get_lifetimeShort(self) -> str:
        lifetime = self.lastUpdateTime - self.timestamp
        hours, remainder = divmod(lifetime.seconds, 3600)
        minutes, seconds = divmod(remainder, 60)
        return f"{hours:02}:{minutes:02}:{seconds:02}"
    
    def get_lifetimeNormalized(self):
        lifetime = self.lastUpdateTime - self.timestamp
        hours, remainder = divmod(lifetime.seconds, 3600)
        minutes, seconds = divmod(remainder, 60)
        minutes_n = self.remap(minutes,0,60,0,1)
        hours_n = self.remap(hours,0,24,0,1)
        return hours_n,minutes_n
    
    def get_lifetimeMinutes(self):
        lifetime = self.lastUpdateTime - self.timestamp
        minutes_n, remainder = divmod(lifetime.seconds, 60)
        return  minutes_n
    
    def get_lifetimeSeconds(self):
        lifetime = self.lastUpdateTime - self.timestamp
        
        return  lifetime.seconds
    
    def close(self, close_price: float):
        if self.status == "Running":
            self.lastPrice=close_price
            self.status = "Closed"
         # print(f"Closed order: {self.to_dict()}")

    def to_dict(self) -> Dict:
        return {
            "symbol": self.symbol,
            "quantity": self.quantity,
            "price": self.price,
            "lastPrice": self.lastPrice,
            "order_type": self.order_type,
            "timestamp": self.timestamp.isoformat(),
            "lastUpdateTime": self.lastUpdateTime.isoformat(),
            "status": self.status,
            "pnl": self.pnl,
            "pnlHigh": self.pnlHigh,
            "pnlLow": self.pnlLow,
            "pnlPercentage": self.pnlPercentage,
            "CloseReason": self.CloseReason,
            "OpenReason": self.OpenReason,
             "UpdateCount": self.UpdateCount,
             "setIndex": self.setIndex,
            "PointsMoved": self.PointsMoved,
            "quantity_Index": self.quantity_Index,
            "groupID": self.groupID,
            "Comment":self.Comment,
            "Take_profit": self.Take_profit,
            "Stop_loss": self.Stop_loss,
              "upside_order": self.upside_order,
                "downside_order": self.downside_order,
                "estimated_points": self.estimated_points,
                "id": self.id
        }

    @classmethod
    def from_dict(cls, data: Dict) -> 'Order':
        order = cls(data['symbol'], data['quantity'], data['price'], data['order_type'], datetime.fromisoformat(data['timestamp']))
        order.status = data['status']
        order.pnl = data['pnl']
        order.pnlPercentage = data['pnlPercentage'] 
        order.lastUpdateTime=datetime.fromisoformat(data['lastUpdateTime'])
        order.lastPrice=data['lastPrice']
        order.pnlLow=data['pnlLow'] 
        order.pnlHigh=data['pnlHigh'] 
        order.CloseReason = data.get('CloseReason', 'Not Found')
        order.OpenReason = data.get('OpenReason', 'Not Found')
        order.UpdateCount = data.get('UpdateCount', 'Not Found')
        order.setIndex = data.get('setIndex', 0)
        order.PointsMoved = data.get('PointsMoved', 0)
        order.quantity_Index = data.get('quantity_Index', 0)
        order.groupID = data.get('groupID', 'Group1')
        order.Comment = data.get('Comment', 'Comment')
        order.Take_profit = data.get('Take_profit', 0)
        order.Stop_loss = data.get('Stop_loss', 0)
        order.upside_order = data.get('upside_order', False)
        order.downside_order = data.get('downside_order', False)
        order.estimated_points = data.get('estimated_points', 0)
        order.id = data.get('id', f"{order.symbol}_{order.timestamp.strftime('%Y%m%d_%H%M%S')}")
        return order

class OrderManager:
    def __init__(self,name="OrderManager"):
        self.orders: List[Order] = []
        self.runningPositionPrice = 0
        self.runningPositionType = ""
        self.runningPnL =0
        self.startTime  = datetime.now()
        self.name=name
        self.can_save = True
        if name == "orderManager":
            print(f"{self.name} is the default OrderManager, please rename it to avoid confusion.")
        

    def place_order(self, symbol: str, quantity: int, price: float, order_type: str, reason:str) -> Optional[Order]:
        if any(order.status == "Running" and order.symbol==symbol for order in self.orders):
            # print(f"{self.name} Cannot open new order. There is already a live order running.")
            return None
        
        new_order = Order(symbol, quantity, price, order_type)
        new_order.OpenReason=reason
        self.orders.append(new_order)
        return new_order
    
    def place_order_direct(self, symbol: str, quantity: int, price: float, order_type: str, reason:str) -> Optional[Order]:
        new_order = Order(symbol, quantity, price, order_type)
        new_order.OpenReason=reason
        self.orders.append(new_order)
        return new_order
    
    def place_order_side_Intented(self, symbol: str, quantity: int, price: float, order_type: str, reason:str) -> Optional[Order]:
        if any(order.status == "Running" and order.symbol==symbol and order.order_type==order_type for order in self.orders):
            return None
        
        new_order = Order(symbol, quantity, price, order_type)
        new_order.OpenReason=reason
        self.orders.append(new_order)
        return new_order

    def close_order(self, order: Order, close_price: float, reason:str):
        if order in self.orders and order.status == "Running":
            order.close(close_price)
            order.CloseReason=reason

    def Update_orders(self, close_price: float , timestamp):
        for order in self.orders:
            if order.status == "Running":
                order.Update(close_price)
                if timestamp != None:
                    order.lastUpdateTime = timestamp
    def Update(self):
        pass

    def get_closed_orders_all(self) -> List[Order]:
        return [order for order in self.orders if order.status != "Running"]
    def get_open_orders_all(self) -> List[Order]:
        return [order for order in self.orders if order.status == "Running"]

    def get_open_orders(self,symbol='BTCUSDT') -> List[Order]:
        return [order for order in self.orders if order.status == "Running" and order.symbol==symbol]
    
    def get_all_orders_symbol(self,symbol='BTCUSDT') -> List[Order]:
        return [order for order in self.orders if order.symbol==symbol]
    def calculate_total_pnl(self) -> float:
        return sum(order.pnl for order in self.orders)
    
    def get_group_pnl(self,symbol='BTCUSDT',groupID = 'Group1') -> float:
        return sum(order.pnl for order in self.orders if  order.symbol==symbol and order.groupID == groupID) 
    
    def get_all_group_orders_symbol(self,symbol='BTCUSDT',groupID = 'Group1') -> List[Order]:
        return [order for order in self.orders if order.symbol==symbol and order.groupID == groupID]
    
    def calculate_total_pnlPercentage(self) -> float:
        return sum(order.pnlPercentage for order in self.orders) 

    def save_orders(self, filename: str =""):
        if self.can_save == False:
            return
        if filename == None or filename == "":
            filename= f'data/{self.name}.json'
        try:
            with open(filename, 'w') as f:
                json.dump([order.to_dict() for order in self.orders], f)
        except IOError as e:
            print(f"{self.name} Error saving orders to file: {e}")
        except Exception as e:
            print(f"{self.name} An unexpected error occurred: {e}")


    def load_orders(self, filename: str =""):
        if filename == None or filename == "":
            filename= f'data/{self.name}.json'
        try:
            with open(filename, 'r') as f:
                data = json.load(f)
                self.orders = [Order.from_dict(order_data) for order_data in data]
        except FileNotFoundError:
            print(f"{self.name} Error: File {filename} not found.")
        except json.JSONDecodeError as e:
            print(f"{self.name} Error: Invalid JSON in file {filename}: {e}")
        except Exception as e:
            print(f"{self.name} An unexpected error occurred: {e}")

    def PrintOrders(self):
        if self.orders:
            print(f"\n\n-----------------------------{self.name} PrintOrders----------------------------------\n")
            headers = ['symbol', 'quantity', 'price', 'lastPrice','PointsMoved', 'order_type', 'timestamp', 'lastUpdateTime', 'lifetime', 'status', 'pnl', 'pnlHigh', 'pnlLow', 'pnlPercentage','Take_profit','Stop_loss']
            table = [[
                getattr(order, header).strftime('%d %A  %I:%M:%S %p') if header in ['timestamp', 'lastUpdateTime']
                else ('+' + str(getattr(order, header)) if getattr(order, header) > 0 else str(getattr(order, header))) if header in ['pnl', 'pnlHigh', 'pnlLow', 'pnlPercentage']
                else order.get_lifetimeShort() if header == 'lifetime'
                else str(getattr(order, header))
                for header in headers
            ] for order in self.orders]

            print(tabulate(table, headers=headers))
            print("\n\n")
            print(f"\n-----------------------------END {self.name} PrintOrders ----------------------------------")

    def get_latest_order(self, symbol='BTCUSDT') -> Optional[Order]:
        symbol_orders = [order for order in self.orders if order.symbol == symbol]
        if not symbol_orders:
            return None  # If there are no orders for the symbol, return None
        latest_order = max(symbol_orders, key=lambda order: order.lastUpdateTime)
        return latest_order

    
    def setRunningInfo(self, symbol="BTCUSDT"):
         open_orders = self.get_open_orders()
         if open_orders:
                 for order in open_orders:
                    self.runningPositionPrice = order.price
                    self.runningPositionType = order.order_type
                    self.runningPnL = order.pnl
                    self.startTime  = order.timestamp
                    self.runlifeTime  =  order.lastUpdateTime - order.timestamp


    def Get_live_positionLimits(self,symbol):
        open_orders = self.get_open_orders(symbol=symbol)
        if open_orders:
            for order in open_orders:
                if(order.order_type=="buy"):
                    return f"\nAlready a Live Long Position is Running ,so Dont Choose OpenLong OR OpenShort or HoldShort.\n "
                if(order.order_type=="sell"):
                     return f"\nAlready a Live Short Position is Running ,so Dont Choose OpenLong OR OpenShort or HoldLong .\n "
        else:
            return "\n"    
        
    def get_running_orderInfo(self, orderIN=None):
        openOrders = self.get_open_orders_all()
        if len(openOrders) != 0 :
            order = openOrders[0]
            if orderIN:
                order=orderIN
            lifetime_hours,lifetime_minutes = order.get_lifetimeNormalized()
            reward = (order.pnlPercentage-0.01)
            if reward <0:
                reward *= 6

            order_type= 0
            if order.order_type == "buy":
                order_type=1
            return reward,order_type,order.pnlHighPercentage,order.pnlLowPercentage,lifetime_hours,lifetime_minutes
        else:
            return 0,0,0,0,0,0
    
    
        
    def get_all_orders_info(self):
        win_count = sum(1 for order in self.orders if  order.pnl > 0)
        lost_count = sum(1 for order in self.orders if order.pnl < 0)
        longs = sum(1 for order in self.orders if  order.order_type =="buy")
        shorts = sum(1 for order in self.orders if  order.order_type !="buy")
        txt=f"w{win_count}_L{lost_count}_LS_{longs}_{shorts}_Total{len(self.orders )}"
        return txt
    

    def show_orders_and_pnl(self,noclose= True , PrintText=True , symbol="BTCUSDT" , onlyPnLInfo = True):
        open_orders = self.get_open_orders(symbol=symbol)
        closed_orders = [order for order in self.orders if order.status == "Closed"]
        self.setRunningInfo(symbol=symbol)
        if open_orders:
            output = f"\n-------------------------- {self.name} Live Running Positions Details -------------------------------\n\n"
            for order in open_orders:
                
                positionType = "Short"
                if order.order_type=="buy":
                    positionType="Long"

                output += f"Timestamp: {order.timestamp.strftime('%I:%M:%S %p')} Symbol: {order.symbol}, Quantity: {order.quantity} , OpenPrice: {round(order.price, 6)} => lastPrice: {round(order.lastPrice, 6)} , "
                output += f", position Type: {order.order_type} {order.status} ,Current Pnl Percentage {order.pnlPercentage}% ,Current Pnl $ {order.pnl} , Pnl High Reached $ {order.pnlHigh} "
                output += f" , lastUpdateTime: {order.lastUpdateTime.strftime('%I:%M:%S %p')} , {order.get_lifetime()}\n "
                output += f"{order.order_type} Position started at {order.price} ,From there Price moved around $ {round(order.lastPrice - order.price, 6)} .Latest Last price is  $ {round(order.lastPrice, 6)}   \n"
                
                if order.pnl > 0 and abs(order.pnlHigh-order.pnl) > 236:
                    output +=f", Pnl High Reached $ {order.pnlHigh} , Pnl Low Reached $ {order.pnlLow}"
                    output +=f"\nCurrent Running {positionType} Position has decresed profits from its position peak high reach $ {order.pnlHigh} is $ {order.pnlHigh-order.pnl} \n"

                if order.pnl > 0 :
                    output +=f"current Running {positionType} Position is in PROFIT of $ {order.pnl} with +{order.pnlPercentage}%\n"
                else :
                    output +=f"current Running {positionType} Position is in LOSS of $ {order.pnl} with {order.pnlPercentage}%\n"


            output += "\n----------------------------------------------------------\n"
        else:
            output = f"\n--------------------------{self.name} No Running Positions-------------------------------\n\n"
            output += "No Running positions exists.so dont choose HoldLong Or HoldShort ,No position is in live Running,nothing  to Hold \n"
            output += "\n----------------------------------------------------------\n"
        
        if noclose == True:
            if closed_orders:
                output += f"\n--------------------------{self.name} Closed Positions-------------------------------\n\n"
                for order in closed_orders:
                    output += f"Timestamp: {order.timestamp.strftime('%I:%M:%S %p')} Symbol: {order.symbol}, Quantity: {order.quantity}, OpenPrice: {round(order.price, 6)} => "
                    output += f" lastPrice: {round(order.lastPrice, 6)}, position Type: {order.order_type} {order.status} , Closed Pnl Percentage $ {order.pnlPercentage}% "
                    output += f" , Closed Pnl $ {order.pnl} , Pnl High Reached $ {order.pnlHigh} , Pnl Low Reached $ {order.pnlLow} , lastUpdateTime: {order.lastUpdateTime.strftime('%I:%M:%S %p')} , {order.get_lifetime()}\n"
                output += "\n----------------------------------------------------------\n"
            else:
                output += f"\n-------------------------- {self.name} Closed Positions -------------------------------\n\n"
                output += "No closed positions exists.\n"
                output += "\n----------------------------------------------------------\n"
        if onlyPnLInfo:
            output=f"\n-------------------------{self.name} PnLInfo-----------------------------\n"
        output += f"{self.name} Total Running Positions: {len(open_orders)}\n"
        output += f"{self.name} Total Closed Positions: {len(closed_orders)}\n"
        output += f"{self.name} Total Trades: {len(self.orders)}\n"
        
        win_count = sum(1 for order in self.orders if  order.pnl > 0)
        lost_count = sum(1 for order in self.orders if order.pnl < 0)
        output += f"{self.name} Win Count: {win_count}\n"
        output += f"{self.name} Lost Count: {lost_count}\n"
        
        total_pnl = self.calculate_total_pnl()
        total_pnl_percentage = self.calculate_total_pnlPercentage()
        output += f"{self.name} Total PnL: $ {'+' if total_pnl > 0 else ''}{total_pnl:.2f}\n"
        output += f"{self.name} Total PnL Rupees: ₹ {'+' if total_pnl*85 > 0 else ''}{(total_pnl*85):.2f}\n"
        output += f"{self.name} Total PnL Percentage: {'+' if total_pnl_percentage > 0 else ''}{total_pnl_percentage:.2f}%\n"
        output += f"---------------------------------------- End {self.name} ----------------------------------------------\n\n"
        if PrintText:
           print(output)
        
        return output




# Example usage
if __name__ == "__main__":
    manager = OrderManager()
    # Place an order
    order1 = manager.place_order("AAPL", 100, 150.0, "buy", reason="Rest")
    # Try to place another order (should fail)
    order2 = manager.place_order("GOOGL", 50, 2500.0, "buy",reason="Rest")
    # Close the first order
    manager.close_order(order1, 160.0,reason="Rest")

    pnl,pnlHigh,pnlLow,lifetime_hours,lifetime_minutes = manager.get_running_orderInfo()

    manager.show_orders_and_pnl(noclose=True)

    print(f"Closed order: {order1.to_dict()}")
    
    # Calculate PNL
    total_pnl = manager.calculate_total_pnl()

    print(f"Total PNL: ${total_pnl}")
    
    # Save orders
    manager.save_orders("IgnoredFloder/ordersTest.json")
    
    # Load orders
    new_manager = OrderManager()
    new_manager.load_orders("Data/orders.json")
    print(f"Loaded {len(new_manager.orders)} orders")

    new_manager.PrintOrders()