Using @mde answer's for the average price method and developing it for Fifo method:
# avg price based PnLCalculator
class PnLCalculator:
def __init__(self):
self.quantity = 0
self.cost = 0.0
self.market_value = 0.0
self.r_pnl = 0.0
self.average_price = 0.0
def fill(self, pos_change, exec_price):
n_pos = pos_change + self.quantity
direction = np.sign(pos_change)
prev_direction = np.sign(self.quantity)
qty_closing = min(abs(self.quantity), abs(pos_change)) * direction if prev_direction != direction else 0
qty_opening = pos_change if prev_direction == direction else pos_change - qty_closing
new_cost = self.cost + qty_opening * exec_price
if self.quantity != 0:
new_cost += qty_closing * self.cost / self.quantity
self.r_pnl += qty_closing * (self.cost / self.quantity - exec_price)
self.quantity = n_pos
self.cost = new_cost
def update(self, price):
if self.quantity != 0:
self.average_price = self.cost / self.quantity
else:
self.average_price = 0
self.market_value = self.quantity * price
return self.market_value - self.cost
Here is a similar class for Fifo and LIFO pnl:
# Fifo or Lifo PnL
class FIFOPnlCalculator:
def __init__(self, isFifo=True):
self.open_trades = []
self.closed_trades = []
self.fifoIndex = -1 if isFifo else 0
self.r_pnl = 0
self.quantity = 0
self.average_price = 0
def pnl(self,d):
return (d['close_price']-d['open_price'])*d['pos']
def fill(self, pos_change, exec_price):
# new trade
if len(self.open_trades)==0:
self.open_trades = [{'pos':pos_change, 'price':exec_price}]
return
last_open = self.open_trades[self.fifoIndex]
# new trade increases position
if last_open['pos']*pos_change>0:
self.open_trades += [{'pos':pos_change, 'price':exec_price}]
return
# new trade smaller or equal than last open trade
if abs(last_open['pos'])>=abs(pos_change):
d = {'pos':-pos_change, 'open_price':last_open['price'], 'close_price':exec_price}
self.closed_trades += [d]
self.r_pnl += self.pnl(d)
last_open['pos'] += pos_change
if last_open['pos']==0:
self.open_trades.pop(self.fifoIndex)
return
# new trade greater exhausts last open trade
d = {'pos':-pos_change, 'open_price':last_open['price'], 'close_price':exec_price}
self.closed_trades += [d]
self.r_pnl += self.pnl(d)
pos_change += last_open['pos']
self.open_trades.pop(self.fifoIndex)
self.fill(pos_change, exec_price)
def update(self, price):
u_pnl = 0
self.quantity = 0
self.average_price = 0
for r in self.open_trades:
u_pnl += r['pos']*(price-r['price'])
self.quantity += r['pos']
self.average_price += r['pos']*r['price']
if self.quantity!=0:
self.average_price /= self.quantity
return u_pnl
Here is the code to test for the results for average price:
quantities = np.array([200, 100, -100, 100, 100, -400])
exec_prices = np.array([50.0, 51.0, 49.0, 51.0, 53.0, 52.0])
pnls = []
print('Pos\t|\tR.P&L\t|\tU P&L\t|\tAvgPrc')
print('-' * 55)
pos = PnLCalculator()
pnls = []
for (p,e) in zip(quantities, exec_prices):
pos.fill(p, e)
u_pnl = pos.update(e)
print('%+d\t|\t%.1f\t|\t%.1f\t|\t[%.1f]' % (pos.quantity, pos.r_pnl, u_pnl, pos.average_price))
pnls.append(u_pnl + pos.r_pnl)
print('-' * 55)
print(pnls)
and for Fifo:
quantities = np.array([200, 100, -100, 100, 100, -400])
exec_prices = np.array([50.0, 51.0, 49.0, 51.0, 53.0, 52.0])
pnls = []
print('Pos\t|\tR.P&L\t|\tU P&L\t|\tAvgPrc')
print('-' * 55)
pos = FIFOPnlCalculator(False)
pnls = []
for (p,e) in zip(quantities, exec_prices):
pos.fill(p, e)
u_pnl = pos.update(e)
print('%+d\t|\t%.1f\t|\t%.1f\t|\t[%.1f]' % (pos.quantity, pos.r_pnl, u_pnl, pos.average_price))
pnls.append(u_pnl + pos.r_pnl)
print('-' * 55)
print(pnls)
As expected, both code output the same pnl, but the breakdown between realized and unrealized pnl is different.
