Cost basis to positions

lumibot/backtesting/backtesting_broker.py

@@ -449,6 +449,29 @@ def process_pending_orders(self, strategy):
  low = ohlc.df["low"][-1]
  close = ohlc.df["close"][-1]
  volume = ohlc.df["volume"][-1]
+
+ elif self._data_source.SOURCE == "REST":
+ timeshift = timedelta(
+ minutes=0
+ )
+ ohlc = strategy.get_historical_prices(
+ asset,
+ 1,
+ quote=order.quote,
+ timeshift=timeshift,
+ timestep=self._data_source._timestep,
+ )
+
+ if ohlc is None:
+ self.cancel_order(order)
+ continue
+ dt = ohlc.df.index[-1]
+ open = ohlc.df["open"][-1]
+ high = ohlc.df["high"][-1]
+ low = ohlc.df["low"][-1]
+ close = ohlc.df["close"][-1]
+ volume = ohlc.df["volume"][-1]
+
 
  # Determine transaction price.
  if order.type == "market":

lumibot/backtesting/data_source_backtesting.py

@@ -61,6 +61,8 @@ def _pull_source_symbol_bars(
  backtesting_timeshift = timeshift
  elif self.LIVE_DATA_SOURCE.SOURCE == "ALPHA_VANTAGE":
  backtesting_timeshift = timeshift
+ elif self.LIVE_DATA_SOURCE.SOURCE == "REST":
+ backtesting_timeshift = timeshift
  else:
  raise ValueError(
  f"An incorrect data source type was received. Received"

lumibot/brokers/alpaca.py

@@ -5,6 +5,7 @@
 from asyncio import CancelledError
 from datetime import timezone
 from decimal import Decimal
+import uuid
 
 import alpaca_trade_api as tradeapi
 from alpaca_trade_api.stream import Stream
@@ -236,8 +237,53 @@ def _parse_broker_position(self, broker_position, strategy, orders=None):
  )
 
  quantity = position["qty"]
- position = Position(strategy, asset, quantity, orders=orders)
+ # Set the cost_basis of the position
+ cost_basis = float(position["cost_basis"])
+ print(f'Alpaca cost basis for {asset} - {cost_basis}')
+
+ # TLNG - Check if this symbol is already managed by our strategy
+ all_known_positions = self._filled_positions + self._untracked_positions
+ if position["symbol"] not in [pos.symbol for pos in all_known_positions]:
+ # If it's not, check if it should be
+ strategy = self._confirm_strategy_for_broker_position(position, strategy)
+
+ position = Position(strategy, asset, quantity, orders=orders, cost_basis=cost_basis)
  return position
+
+ def _confirm_strategy_for_broker_position(self, position, strategy):
+ print(f'Dealing with {position["symbol"]}')
+ qty_left = float(position['qty'])
+ order_ids = []
+ activities = self.api.get_activities(activity_types='FILL', page_size=100)
+
+ while True:
+ for activity in activities:
+ if qty_left == 0.0:
+ break
+ if activity.symbol == position['symbol']:
+ qty = float(activity.qty) if activity.side == 'buy' else -float(activity.qty)
+ # print(f'We found an activity for {position.symbol} - qty: {qty} - {activity.side} @ {activity.transaction_time}')
+ qty_left -= qty
+ order_ids.append(activity.order_id)
+ if qty_left != 0.0:
+ print(f'Were still looking for {qty_left} of {position["symbol"]}')
+ activities = self.api.get_activities(activity_types='FILL', page_token=activity.id, page_size=100)
+ # time.sleep(0.5)
+ else:
+ break
+
+ for order_id in order_ids:
+ order = self.api.get_order(order_id=order_id)
+ client_order_id_split = order.client_order_id.split(':')
+ if len(client_order_id_split) > 0:
+ if client_order_id_split[0] == strategy:
+ return strategy
+
+ return 'UNTRACKED'
+
+
+
+
 
  def _pull_broker_position(self, asset):
  """Given a asset, get the broker representation
@@ -313,6 +359,10 @@ def _submit_order(self, order):
  if order.time_in_force != "gtc" or "ioc":
  order.time_in_force = "gtc"
 
+ # Generate a client_order_id that starts with the strategy name and after that has a uid. Max 48 characters
+ # as per Alpacas spec.
+ client_order_id = (order.strategy + ":" + str(uuid.uuid4()))[0:48]
+
  kwargs = {
  "type": order.type,
  "order_class": order.order_class,
@@ -321,6 +371,9 @@ def _submit_order(self, order):
  "stop_price": str(order.stop_price) if order.stop_price else None,
  "trail_price": str(order.trail_price) if order.trail_price else None,
  "trail_percent": order.trail_percent,
+ "client_order_id": client_order_id, # TLNG
+ "extended_hours": True, # TODO: Remove this line!
+ "time_in_force": "day", # TODO: Remove this line!
  }
  # Remove items with None values
  kwargs = {k: v for k, v in kwargs.items() if v}

lumibot/brokers/broker.py

@@ -37,6 +37,7 @@ def __init__(self, name="", connect_stream=True):
  self._canceled_orders = SafeList(self._lock)
  self._partially_filled_orders = SafeList(self._lock)
  self._filled_positions = SafeList(self._lock)
+ self._untracked_positions = SafeList(self._lock) # TLNG - used to keep track of positions that we DON'T care about
  self._subscribers = SafeList(self._lock)
  self._is_stream_subscribed = False
  self._trade_event_log_df = pd.DataFrame()
@@ -117,7 +118,10 @@ def _set_initial_positions(self, strategy):
  """ Set initial positions """
  positions = self._pull_positions(strategy)
  for pos in positions:
- self._filled_positions.append(pos)
+ if strategy.name == pos.strategy: # TLNG - Only add the position if it belongs to our strategy
+ self._filled_positions.append(pos)
+ else:
+ self._untracked_positions.append(pos)
 
  def _process_new_order(self, order):
  logging.info(colored("New %r was submitted." % order, color="green"))

lumibot/entities/position.py

@@ -1,7 +1,18 @@
 from decimal import Decimal, getcontext
-
 import lumibot.entities as entities
 
+from collections import deque
+from dataclasses import dataclass
+
+@dataclass
+class MutableTrans:
+ '''
+ This is just a convenience class to use in the cost_basis_calculation. It contains the same data as the
+ Transactions named tuple from order.py, but it is mutable which makes it easier to use.
+ '''
+ quantity: float
+ price: float
+
 
 class Position:
  """
@@ -22,7 +33,7 @@ class Position:
  The orders that have been executed for this position.
  """
 
- def __init__(self, strategy, asset, quantity, orders=None, hold=0, available=0):
+ def __init__(self, strategy, asset, quantity, orders=None, hold=0, available=0, cost_basis=0.0):
  self.strategy = strategy
  self.asset = asset
  self.symbol = self.asset.symbol
@@ -56,6 +67,16 @@ def __init__(self, strategy, asset, quantity, orders=None, hold=0, available=0):
  )
  self.orders = orders
 
+ # cost_basis is the amount of money it took to aquire this position
+ # it can be derived from summarising all orders, or can be aquired from Alpaca
+ self.cost_basis = cost_basis
+ self.calculate_cost_basis_from_orders = False
+
+ # If we didn't receive cost basis from the broker, try to calculate it from the orders.
+ if self.cost_basis == 0.0:
+ self.calculate_cost_basis_from_orders = True
+ self.update_cost_basis_from_orders()
+
  def __repr__(self):
  repr = "%f shares of %s" % (self.quantity, self.asset)
  return repr
@@ -74,6 +95,10 @@ def quantity(self):
  def quantity(self, value):
  self._quantity = Decimal(value)
 
+ @property
+ def avg_entry_price(self):
+ return self.cost_basis / self.quantity if self.quantity else 0.0
+
  @property
  def hold(self):
  return self._hold
@@ -156,3 +181,50 @@ def add_order(self, order: entities.Order, quantity: Decimal):
  self._quantity += Decimal(increment)
  if order not in self.orders:
  self.orders.append(order)
+
+ # Update cost_basis to include this order as well
+ if self.calculate_cost_basis_from_orders:
+ self.update_cost_basis_from_orders()
+
+
+ def update_cost_basis_from_orders(self):
+ ''' Update positions cost_basis based on available orders and their transactions. '''
+
+ # Separate all transactions in buys and sells
+ buys = deque()
+ sells = deque()
+ for order in self.orders:
+ for transaction in order.transactions:
+ qty = float(transaction.quantity)
+ qty = qty if order.side == 'buy' else -qty
+ print(f'Cost_basis {order.asset}: {order} - qty: {qty} price: {transaction.price}')
+ if qty > 0.0:
+ buys.append(MutableTrans(quantity=qty, price=transaction.price))
+ elif qty < 0.0:
+ sells.append(MutableTrans(quantity=qty, price=transaction.price))
+
+ # Emulate FIFO to determine cost basis
+ # Loop all buys/sells until one of the lists run out
+ while True:
+ if len(buys) == 0 or len(sells) == 0:
+ break
+ diff = buys[0].quantity - abs(sells[0].quantity)
+ if diff > 0.0:
+ sells.popleft()
+ buys[0].quantity = diff
+ elif diff < 0.0:
+ buys.popleft()
+ sells[0].quantity = diff
+ else:
+ sells.popleft()
+ buys.popleft()
+
+ # After FIFOing all transactions, what we have left are the shares that makes up the cost basis.
+ cost_price = 0.0
+ total_qty = 0.0
+ for transaction in list(buys + sells):
+ cost_price += transaction.quantity * transaction.price
+ total_qty += transaction.quantity
+
+ self.cost_basis = cost_price
+ print(f'Cost_basis updated: {cost_price}')

lumibot/strategies/strategy_executor.py

@@ -136,6 +136,10 @@ def func_output(self, *args, **kwargs):
  # Compare to existing lumi position.
  if position_lumi.quantity != position.quantity:
  position_lumi.quantity = position.quantity
+
+ # Update the positions cost_basis with the data from the broker
+ # TODO: Check so that this works with other brokers than Alpaca.
+ position_lumi.cost_basis = position.cost_basis
  else:
  # Add to positions in lumibot, position does not exist
  # in lumibot.