Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from lumibot.brokers import Alpaca
- from lumibot.backtesting import YahooDataBacktesting
- from lumibot.strategies.strategy import Strategy
- from lumibot.traders import Trader
- from datetime import datetime, timedelta
- from alpaca_trade_api import REST
- from timedelta import Timedelta
- API_KEY = # ALPACA API KEY
- API_SECRET = # ALPACA API SECRET
- BASE_URL = "https://paper-api.alpaca.markets"
- ALPACA_CREDS = {
- "API_KEY" : API_KEY,
- "API_SECRET" : API_SECRET,
- "PAPER" : True
- }
- class BuyTheDip(Strategy):
- def initialize(self, symbol:str="SPY", cash_at_risk:float=.5, close_distance_threshold:float=20):
- self.symbol = symbol
- self.sleeptime = "24H"
- self.last_trade = None
- self.cash_at_risk = cash_at_risk
- self.close_distance_threshold = close_distance_threshold
- self.api = REST(base_url=BASE_URL, key_id=API_KEY, secret_key=API_SECRET)
- def position_sizing(self):
- cash = self.get_cash()
- last_price = self.get_last_price(self.symbol)
- quantity = round(cash * self.cash_at_risk / last_price, 0)
- return cash, last_price, quantity
- def get_previous_day_data(self):
- today = self.get_datetime()
- yesterday = today - timedelta(days=1)
- bars = self.api.get_bars(self.symbol, '1Day', start=yesterday.strftime('%Y-%m-%d'), end=today.strftime('%Y-%m-%d'), limit=1).df
- if len(bars) > 0:
- return bars.iloc[0]
- return None
- def calculate_close_distance_percentage(self, bar):
- if bar is None:
- return None
- day_range = bar['high'] - bar['low']
- close_distance = bar['close'] - bar['low']
- close_distance_percentage = (close_distance / day_range) * 100
- return close_distance_percentage
- def on_trading_iteration(self):
- cash, last_price, quantity = self.position_sizing()
- previous_day_bar = self.get_previous_day_data()
- close_distance_percentage = self.calculate_close_distance_percentage(previous_day_bar)
- if close_distance_percentage is None:
- self.log_message("Unable to calculate close distance percentage. Skipping trade.")
- return
- if cash > last_price:
- if close_distance_percentage <= self.close_distance_threshold:
- if self.last_trade == "sell":
- self.sell_all()
- order = self.create_order(
- self.symbol,
- quantity,
- "buy",
- type="bracket",
- take_profit_price=last_price*1.02,
- stop_loss_price=last_price*.98
- )
- self.submit_order(order)
- self.last_trade = "buy"
- self.log_message(f"Bought {quantity} shares of {self.symbol} at {last_price}. Close distance: {close_distance_percentage:.2f}%")
- else:
- self.log_message(f"No trade. Close distance: {close_distance_percentage:.2f}% is above threshold of {self.close_distance_threshold}%")
- def run_strategy(symbol:str="SPY", cash_at_risk:float=0.5, close_distance_threshold:float=20,
- start_date:datetime=None, end_date:datetime=None, live:bool=False):
- broker = Alpaca(ALPACA_CREDS)
- strategy = BuyTheDip(name='rangestrat', broker=broker,
- parameters={"symbol": symbol,
- "cash_at_risk": cash_at_risk,
- "close_distance_threshold": close_distance_threshold})
- if live:
- trader = Trader()
- trader.add_strategy(strategy)
- trader.run_all()
- else:
- strategy.backtest(
- YahooDataBacktesting,
- start_date,
- end_date,
- parameters={"symbol": symbol, "cash_at_risk": cash_at_risk, "close_distance_threshold": close_distance_threshold}
- )
- run_strategy(
- symbol="SPY",
- cash_at_risk=0.5,
- close_distance_threshold=20,
- start_date=datetime(2024,1,1),
- end_date=datetime(2024,8,1),
- live=False
- )
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement