Created
September 28, 2024 06:47
-
-
Save spchamp/e2361ee3e7b74f7c276ebd461b20ce6e to your computer and use it in GitHub Desktop.
main.py for quantconnect (needs debug, backtest failing for missing data)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # region imports | |
| from datetime import timedelta | |
| from enum import StrEnum | |
| import pandas as pd | |
| from typing import TYPE_CHECKING, Any | |
| from AlgorithmImports import ( | |
| QCAlgorithm, Resolution, Slice, Market, TimeSpan, | |
| MovingAverageType, TradeBarConsolidator, SubscriptionManager, | |
| TradeBar, BollingerBands, BaseDataConsolidator | |
| ) | |
| import numpy as np | |
| # endregion | |
| # after | |
| # https://algotrading101.com/learn/quantconnect-guide/ | |
| # https://github.com/GregBland/QuantConnect_article/blob/main/leanHogsAlgo.py | |
| SYMBOL: str = "NZDUSD" | |
| TEST_ORDER_SIZE: int = 100 | |
| BB_PERIOD: int = 15 | |
| BB_K = 1.5 | |
| BB_MA = MovingAverageType.WILDERS | |
| BB_RESOL = Resolution.MINUTE | |
| class Const(StrEnum): | |
| SYMBOL = "symbol" | |
| WARMUP = "warmup" | |
| INDICATOR = "indicator" | |
| class LeanTest00(QCAlgorithm): | |
| initialized: bool = False | |
| if TYPE_CHECKING: | |
| params: dict[str, Any] | |
| def __init__(self) -> None: | |
| super().__init__() | |
| self.params = dict() | |
| def initialize(self): | |
| global SYMBOL, WARMUP_DELTA | |
| self.set_start_date(2024, 6, 1) # Set Start Date | |
| self.set_end_date(2024, 8, 31) # Set End Date | |
| self.set_cash(2000) # Set Strategy Cash | |
| fx = self.add_forex(SYMBOL, resolution=Resolution.MINUTE, | |
| market=Market.OANDA, leverage=50) | |
| symbol = fx.symbol | |
| self.log("-- symbol %r" % symbol) | |
| consol_15m = self.create_consolidator(np.timedelta64( | |
| 15, "m").astype("O"), TradeBarConsolidator) | |
| consol_15m.data_consolidated += self.on_15m | |
| mng: SubscriptionManager = self.subscription_manager | |
| mng.add_consolidator(symbol, consol_15m) | |
| self.log("-- Added consolidator %r" % consol_15m) | |
| bb = self.bb(symbol, BB_PERIOD, BB_K, BB_MA, BB_RESOL) | |
| self.params[Const.INDICATOR] = bb | |
| self.params[Const.SYMBOL] = symbol | |
| def on_data(self, data: Slice): | |
| symbol = self.params[Const.SYMBOL] | |
| bar: TradeBar = data.bars.get(symbol) | |
| bb: BollingerBands = self.params[Const.INDICATOR] | |
| if bar: | |
| bb.update(bar.EndTime, np.mean( | |
| [bar.high, bar.low, bar.close, bar.close])) | |
| if bb.is_ready: | |
| self.plot("BollingerBands", "bb", bb.current.value) | |
| self.plot("BollingerBands", "standard_deviation", | |
| bb.standard_deviation.current.value) | |
| self.plot("BollingerBands", "middle_band", | |
| bb.middle_band.current.value) | |
| self.plot("BollingerBands", "upper_band", | |
| bb.upper_band.current.value) | |
| self.plot("BollingerBands", "lower_band", | |
| bb.lower_band.current.value) | |
| self.plot("BollingerBands", "band_width", | |
| bb.band_width.current.value) | |
| self.plot("BollingerBands", "percent_b", | |
| bb.percent_b.current.value) | |
| self.plot("BollingerBands", "price", bb.price.current.value) | |
| def on_15m(self, sender: BaseDataConsolidator, bar: TradeBar): | |
| # event handler for M15 consolidator | |
| if self.is_warming_up: | |
| self.log("Consolidator [warmup] %r " % (bar.time,)) | |
| return | |
| self.log("Consolidator %r" % (bar.time,)) | |
| bb: BollingerBands = self.params[Const.INDICATOR] | |
| if bb.is_ready: | |
| price = np.mean([bar.high, bar.low, bar.close, bar.close]) | |
| if price < bb.lower_band.current.value: | |
| symbol = self.params[Const.SYMBOL] | |
| self.sell(symbol, TEST_ORDER_SIZE) | |
| elif price > bb.upper_band.current.value: | |
| symbol = self.params[Const.SYMBOL] | |
| self.liquidate(symbol) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment