strat-playground/utils/utils.py

from collections import defaultdict, deque
from datetime import datetime
from typing import List

import numpy as np

from internal_types.types import Instrument, Position, Quote

TRADING_DAYS = 252


def sign(x):
  # return x // abs(x)
  # e.g. sign(-5) = -1, sign(0) = 0, sign(5) = 1
  return (x > 0) - (x < 0)


def max_abs(*xs):
  return max(map(abs, xs))


def min_abs(*xs):
  return min(map(abs, xs))


def simple_sharpe_ratio(historical_net_liquid_value: List[float], interval_sec: int) -> float:
  if len(historical_net_liquid_value) < 2:
    return np.nan

  xs = np.asarray(historical_net_liquid_value, dtype=float)
  returns = xs[1:] / xs[:-1] - 1.0
  mean_r = np.mean(returns)
  std_r = np.std(returns, ddof=1)

  if std_r == 0:
    return np.nan

  periods_per_year = TRADING_DAYS * 24 * 60 * 60 / interval_sec

  return (mean_r / std_r) * np.sqrt(periods_per_year)


def log_sharpe_ratio(historical_net_liquid_value: List[float], interval_sec: int) -> float:
  if len(historical_net_liquid_value) < 2:
    return np.nan

  xs = np.asarray(historical_net_liquid_value, dtype=float)
  log_returns = np.log(xs[1:] / xs[:-1])
  mean_r = log_returns.mean()
  std_r = log_returns.std(ddof=1)

  if std_r == 0:
    return np.nan

  periods_per_year = TRADING_DAYS * 24 * 60 * 60 / interval_sec

  return (mean_r / std_r) * np.sqrt(periods_per_year)


# def consolidate_positions(ps: List[Position]) -> Position:
#   # assumes that ps is not empty and contains the same instruments
#   qty = 0
#   outstanding_balance = 0
#   for p in ps:
#     qty += p.quantity
#     outstanding_balance += p.price * p.quantity
#   return Position(ps[0].instr, qty, outstanding_balance / qty if qty else 0)


def timestamp_to_str(timestamp: int) -> str:
  return datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S UTC')


def interval_idx(timestamp: int, interval_sec: int) -> int:
  return timestamp // interval_sec


def long(pos: Position) -> bool:
  return pos.quantity > 0


def unrealized_gains(pos: Position, closing_price: float) -> float:
  return (closing_price - pos.price) * pos.quantity * pos.instr.multiplier


class Portfolio:

  def __init__(self):
    self._realized_gains = defaultdict(float)
    self.positions = defaultdict(deque[Position])

  def empty(self, instr: Instrument) -> bool:
    return len(self.positions[instr]) == 0

  def add_pos(self, new_pos: Position):
    if new_pos.quantity == 0:
      return

    instr = new_pos.instr

    if self.empty(instr) or long(self.positions[instr][0]) == long(new_pos):
      self.positions[instr].append(new_pos)
      return

    while not self.empty(instr) and new_pos.quantity:
      old_pos = self.positions[instr].popleft()
      min_abs_qty = min_abs(old_pos.quantity, new_pos.quantity)
      tmp_pos = Position(instr, min_abs_qty * sign(old_pos.quantity), old_pos.price)
      self._realized_gains[instr] += unrealized_gains(tmp_pos, new_pos.price)
      old_pos.quantity -= min_abs_qty * sign(old_pos.quantity)
      new_pos.quantity -= min_abs_qty * sign(new_pos.quantity)

      if old_pos.quantity:
        self.positions[instr].appendleft(old_pos)

    if new_pos.quantity:
      self.positions[instr].append(new_pos)

  def curr_position(self, instr: Instrument) -> int:
    # todo: handle fractional shares (crypto)
    return sum(pos.quantity for pos in self.positions[instr])

  def realized_gains(self, instr: Instrument) -> float:
    return self._realized_gains[instr]

  def unrealized_gains(self, instr: Instrument, closing_price: float) -> float:
    return sum(unrealized_gains(pos, closing_price) for pos in self.positions[instr])

  def total_gains(self, instr: Instrument, closing_price: float) -> float:
    return self._realized_gains[instr] + self.unrealized_gains(instr, closing_price)


class BlendedCandlesticks:

  def __init__(self, interval_sec: int):
    self.interval_sec = interval_sec
    self.timestamps: List[int] = []
    self.opens: List[float] = []
    self.highs: List[float] = []
    self.lows: List[float] = []
    self.closes: List[float] = []
    self.volumes: List[int] = []  # todo: float for crypto
    self.incomplete_bar: Quote | None = None

  def __len__(self):
    return len(self.timestamps)

  def __append(self, quote: Quote):
    if self.incomplete_bar is not None:
      self.timestamps.append(self.incomplete_bar.timestamp)
      self.opens.append(self.incomplete_bar.open)
      self.highs.append(self.incomplete_bar.high)
      self.lows.append(self.incomplete_bar.low)
      self.closes.append(self.incomplete_bar.close)
      self.volumes.append(self.incomplete_bar.volume)
    self.incomplete_bar = quote

  def __blend(self, quote: Quote):
    if self.incomplete_bar is None:
      self.incomplete_bar = quote
    else:
      self.incomplete_bar = Quote(timestamp=self.incomplete_bar.timestamp,
                                  open=self.incomplete_bar.open,
                                  high=max(self.incomplete_bar.high, quote.high),
                                  low=min(self.incomplete_bar.low, quote.low),
                                  close=quote.close,
                                  volume=self.incomplete_bar.volume + quote.volume)

  def __to_blend(self, quote: Quote):
    if not self.timestamps or self.incomplete_bar is None:
      return False
    last_interval_idx = interval_idx(self.incomplete_bar.timestamp, self.interval_sec)
    quote_interval_idx = interval_idx(quote.timestamp, self.interval_sec)
    return last_interval_idx == quote_interval_idx

  def append(self, quote: Quote):
    if self.__to_blend(quote):
      self.__blend(quote)
    else:
      self.__append(quote)

  def blended_quote(self, idx: int) -> Quote:
    # todo: throw runtime error if idx >= len(self)
    return Quote(timestamp=self.timestamps[idx],
                 open=self.opens[idx],
                 high=self.highs[idx],
                 low=self.lows[idx],
                 close=self.closes[idx],
                 volume=self.volumes[idx])