Skip to content

Instantly share code, notes, and snippets.

@raposatech
Last active December 25, 2023 00:03
Show Gist options
  • Save raposatech/f6b0de855fdb863c9510ef5c85bbe917 to your computer and use it in GitHub Desktop.
Save raposatech/f6b0de855fdb863c9510ef5c85bbe917 to your computer and use it in GitHub Desktop.
Starter System with Alpaca Integration
class AlpacaStarterSystem(DiversifiedStarterSystem):
'''
Carver's Starter System without stop losses, multiple entry rules,
a forecast for position sizing and rebalancing, and multiple instruments.
Adapted from Rob Carver's Leveraged Trading: https://amzn.to/3C1owYn
Allows live trading via the Alpaca API.
DiversifiedStarterSystem found here:
https://gist.github.com/raposatech/d3f10df41c8745b00cb608bd590a986d
'''
def __init__(self, tickers: list, signals: dict,
base_url: str,
target_risk: float = 0.12, margin_cost: float = 0.04,
short_cost: float = 0.001, interest_on_balance: float = 0.0,
start: str = '2000-01-01', end: str = '2020-12-31',
shorts: bool = True, weights: list = [],
max_forecast: float = 2, min_forecast: float = -2,
exposure_drift: float = 0.1, max_leverage: float = 3,
*args, **kwargs):
self.tickers = tickers
self.n_instruments = len(tickers)
self.signals = signals
self.target_risk = target_risk
self.shorts = shorts
self.start = start
self.end = end
self.margin_cost = margin_cost
self.short_cost = short_cost
self.interest_on_balance = interest_on_balance
self.daily_iob = (1 + self.interest_on_balance) ** (1 / 252)
self.daily_margin_cost = (1 + self.margin_cost) ** (1 / 252)
self.daily_short_cost = self.short_cost / 360
self.max_forecast = max_forecast
self.min_forecast = min_forecast
self.max_leverage = max_leverage
self.exposure_drift = exposure_drift
self.signal_names = []
self.weights = weights
self.idm_dict = {
1: 1,
2: 1.15,
3: 1.22,
4: 1.27,
5: 1.29,
6: 1.31,
7: 1.32,
8: 1.34,
15: 1.36,
25: 1.38,
30: 1.4
}
self.base_url = base_url
self._parseBars = np.vectorize(self.__parseBars)
self.max_time = self._getMaxTime()
self.instrument_index = {t: i
for i, t in enumerate(self.tickers)}
def alpacaInit(self, api_key, api_secret, backtest=False):
self.api = tradeapi.REST(api_key, api_secret, self.base_url)
self._setWeights()
self._setIDM()
self._getAlpacaData(backtest)
if 'CAR' in self.signals.keys():
# Add dividends from YFinance
self._getYFData(True)
self._getPortfolio()
self._calcSignals()
self._calcTotalSignal()
def _getMaxTime(self):
max_time = 0
for v in self.signals.values():
for v1 in v.values():
for k, v2 in v1.items():
if k != 'scale' and v2 > max_time:
max_time = v2
return max_time
def __parseBars(self, bar):
return bar.t.date(), bar.c
def _getYFData(self, dividends_only=False):
yfObj = yf.Tickers(self.tickers)
df = yfObj.history(start=self.start, end=self.end)
df.drop(['High', 'Open', 'Stock Splits', 'Volume', 'Low'],
axis=1, inplace=True)
df.columns = df.columns.swaplevel()
if dividends_only:
divs = df.loc[:, (slice(None), 'Dividends')]
divs.index.rename(self.data.index.name, inplace=True)
df = self.data.merge(divs, how='outer',
left_index=True, right_index=True)
data = df.fillna(0)
self.data = data.copy()
def _getPortfolio(self):
account = self.api.get_account()
self.cash = float(account.cash)
# Get current positions
self.current_positions = self.api.list_positions()
self.portfolio_value = np.dot(
self.data.loc[:, (slice(None), 'Close')].iloc[-1].values,
self._getPositions()) + self.cash
self.current_portfolio_value = sum([float(i.market_value)
for i in self.current_positions]) + self.cash
def _getAlpacaData(self, backtest):
'''
Backtests with Alpaca data are not recommended because data pulls
are limited to 1000 bars.
'''
if backtest:
limit = (datetime.today() - pd.to_datetime(self.start)).days
else:
limit = self._getMaxTime()
if limit > 1000:
warn(f'Alpaca data is limited to 1,000 bars. {limit} bars requested.')
limit = 1000
bars = self.api.get_barset(
self.tickers, 'day', limit=limit)
data = pd.DataFrame()
for t in self.tickers:
time, close = self._parseBars(bars[t])
df = pd.DataFrame(close, index=time)
data = pd.concat([data, df], axis=1)
midx = pd.MultiIndex.from_arrays(
[self.tickers, self.n_instruments*['Close']])
data.columns = midx
self.data = data
self.start = data.index[0]
self.end = data.index[-1]
def trade(self):
row = self.data.iloc[-1]
prices = row.loc[(slice(None), 'Close')].values
sigs = row.loc[(slice(None), 'signal')].values
stds = row.loc[(slice(None), 'STD')].values
positions = self._getPositions()
new_pos, cash, shares, delta_exp = self._processBar(
prices, sigs, stds, positions.copy(), self.cash)
orders = self._executeMarketOrder(new_pos - positions)
def _getPositions(self):
positions = np.zeros(self.n_instruments)
for p in self.current_positions:
try:
idx = self.instrument_index[p.symbol]
except KeyError:
# Occurs if there are non-system instruments in the account
continue
positions[idx] += int(p.qty)
return positions
def _executeMarketOrder(self, shares):
print(f"\nPurchase Shares: {shares}")
orders = []
for s, t in zip(shares, self.tickers):
if s > 0:
side = 'buy'
elif s < 0:
side = 'sell'
else:
continue
order = self.api.submit_order(
symbol=t, qty=np.abs(s), side=side,
type='market', time_in_force='gtc')
print(f"{order.side} {order.qty} shares of {order.symbol}")
orders.append(order)
return orders
def backtestInit(self, source='yahoo', starting_capital=10000):
'''Ability to run backtest with this class as well.'''
self.starting_capital = starting_capital
self._getData(source, backtest=True)
self._calcSignals()
self._setWeights()
self._calcTotalSignal()
self._setIDM()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment