| Overall Statistics |
|
Total Orders 1089 Average Win 1.74% Average Loss -1.52% Compounding Annual Return 20.244% Drawdown 21.900% Expectancy 0.360 Start Equity 1000000 End Equity 21002886.28 Net Profit 2000.289% Sharpe Ratio 0.847 Sortino Ratio 0.887 Probabilistic Sharpe Ratio 31.300% Loss Rate 37% Win Rate 63% Profit-Loss Ratio 1.15 Alpha 0.1 Beta 0.236 Annual Standard Deviation 0.154 Annual Variance 0.024 Information Ratio 0.001 Tracking Error 0.198 Treynor Ratio 0.552 Total Fees $93400.93 Estimated Strategy Capacity $8000000000.00 Lowest Capacity Asset UB YYFADOG4CMXX Portfolio Turnover 19.27% Drawdown Recovery 522 |
from AlgorithmImports import *
from datetime import timedelta, datetime
from datetime import time as dt_time
import pandas as pd
import pandas_ta as ta
class M2K_Strategy:
def __init__(self, algorithm):
self.algorithm = algorithm
# ─── MNQ Future ───────────────────────────────────────────────────── Futures.Indices.MICRO_RUSSELL_2000_E_MINI
self.mes_future = self.algorithm.add_future(Futures.Indices.MICRO_RUSSELL_2000_E_MINI, Resolution.MINUTE, extended_market_hours=False)
self.mes_future.set_filter(30, 182)
self.mes_canonical = self.mes_future.symbol
# Managment Varibles
self.mes_position_exposure = 1
self.mes_max_days = 7
self.diagnostical_loging = True
# Empty Varibles
self.current_intraday_bar = None
self.last_date = None
self.entry_time = None
self.active_mes_contract = None
self.regime = None
self.history_bars = []
if self.diagnostical_loging:
self.algorithm.debug("M2K strategy inicializována")
self.algorithm.schedule.on(self.algorithm.date_rules.every_day(),
self.algorithm.time_rules.before_market_close("SPY", 0),
self._execute_trading)
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day('SPY'),
self.algorithm.time_rules.every(timedelta(minutes=60)),
self._loging)
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day('SPY'),
self.algorithm.time_rules.every(timedelta(minutes=5)),
self._loging_before_trading)
def on_warm_up_finished(self):
self.algorithm.debug(f"on warm up finished in M2K MODULE")
self._recover_after_restart()
self._loging()
def _recover_after_restart(self):
for holding in self.algorithm.portfolio.values():
symbol = holding.symbol
if holding.invested:
if "M2K" in symbol.value:
self.active_mes_contract = symbol
holding = self.algorithm.portfolio[symbol]
self.regime = "long" if holding.quantity > 0 else "short"
key = f"entry_time_{symbol.value}"
try:
saved_time = self.algorithm.object_store.read(key)
self.entry_time = datetime.fromisoformat(saved_time)
except:
self.entry_time = self.algorithm.time
self.algorithm.debug(f"Recovered position: {symbol} | Regime: {self.regime} | Entry: {self.entry_time}| Qty: {holding.quantity}")
return
self.algorithm.debug("Not invested in M2K, nothing to recover for M2K_Module")
def on_data(self, slice: Slice):
# Aktualizace mapped futures kontraktu
if self.mes_canonical in slice.future_chains:
future_chain = slice.future_chains[self.mes_canonical]
if future_chain:
sorted_valid = sorted(future_chain, key=lambda c: c.expiry)
choosed_future = sorted_valid[0].symbol
if choosed_future != self.active_mes_contract:
if self.active_mes_contract and self.algorithm.portfolio[self.active_mes_contract].invested:
return
self.active_mes_contract = choosed_future
if not self.algorithm.is_warming_up:
self.algorithm.debug(f"New M2K mapped contract: {choosed_future.value} exp: {sorted_valid[0].expiry.date()}")
# Aktualizace Daily baru
if self.mes_canonical in slice.bars:
data_slice = slice.bars[self.mes_canonical]
t = data_slice.time.time()
d = data_slice.time.date()
#if t < time(9, 30) or t > time(16, 00):
# return
if self.last_date != d:
# Save todays ended bar to history
if self.current_intraday_bar is not None:
self.history_bars.append(self.current_intraday_bar)
if len(self.history_bars) > 250:
self.history_bars.pop(0)
# Creation of new bar for today
self.current_intraday_bar = TradeBar(
data_slice.time, self.mes_canonical,
data_slice.open, data_slice.high, data_slice.low,
data_slice.close, data_slice.volume
)
self.last_date = d
else:
# Actualization of todays bar
bar = self.current_intraday_bar
bar.high = max(bar.high, data_slice.high)
bar.low = min(bar.low, data_slice.low)
bar.close = data_slice.close
bar.volume += data_slice.volume
def _loging(self):
if self.algorithm.is_warming_up or self.active_mes_contract is None:
return
t = self.algorithm.time.time()
if t < time(9, 30) or t > time(16, 00):
return
bar = self.current_intraday_bar
# ─── Calculation of indiacators from todays bar and history ───────────────────────
all_bars = self.history_bars + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
self.algorithm.debug(f"Not Enought Data History M2K + Today bar for Loging | Time: {self.algorithm.time} | Bars: {len(all_bars)}| Today Bar High: {bar.high} | Open: {bar.open} | Low: {bar.low} | Price: {bar.close} ")
return
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_low = self.history_bars[-1].low if self.history_bars else None
invested = self.algorithm.portfolio[self.active_mes_contract].invested
current_price = self.algorithm.securities[self.mes_canonical].price
contract = self.active_mes_contract
ema_10 = self._ema(all_bars, 10, 'close')
sma_200 = self._sma(all_bars, 200, 'close')
self.algorithm.debug(f"MES MODULE -------- {self.algorithm.time} ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------")
self.algorithm.debug(f"Contract: {contract} | Invested: {invested} | SMA 200: {sma_200:.2f}")
self.algorithm.debug(f"Price: {current_price:.2f} | Yesterday close: {yest_close:.2f}")
self.algorithm.debug(f"Bar Close: {bar.close:.2f} | Bar High: {bar.high:.2f} | Bar Low: {bar.low:.2f} | Bar Open: {bar.open:.2f}")
def _loging_before_trading(self):
if self.algorithm.is_warming_up or self.active_mes_contract is None:
return
t = self.algorithm.time.time()
if t < time(15, 40) or t > time(16, 00):
return
bar = self.current_intraday_bar
# ─── Calculation of indiacators from todays bar and history ───────────────────────
all_bars = self.history_bars + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
self.algorithm.debug(f"Not Enought Data History MK2 + Today bar for Loging | Time: {self.algorithm.time} | Bars loaded: {len(all_bars)}| Today Bar High: {bar.high} | Open: {bar.open} | Low: {bar.low} | Price: {bar.close} ")
return
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_low = self.history_bars[-1].low if self.history_bars else None
current_price = self.algorithm.securities[self.mes_canonical].price
invested = self.algorithm.portfolio[self.active_mes_contract].invested
contract = self.active_mes_contract
sma_200 = self._sma(all_bars, 200, 'close')
self.algorithm.debug(f"M2K MODULE -------- {self.algorithm.time} ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------")
self.algorithm.debug(f"Contract: {contract} | Invested: {invested} | SMA 200: {sma_200:.2f}")
self.algorithm.debug(f"Price: {current_price:.2f} | Yesterday close: {yest_close:.2f}")
self.algorithm.debug(f"Bar Close: {bar.close:.2f} | Bar High: {bar.high:.2f} | Bar Low: {bar.low:.2f} | Bar Open: {bar.open:.2f}")
def _execute_trading(self):
if self.algorithm.is_warming_up or self.active_mes_contract is None:
return
bar = self.current_intraday_bar
all_bars = self.history_bars + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
self.algorithm.debug(f"Not Enought Data History M2K + Today bar for Loging | Time: {self.algorithm.time} | Bars loaded: {len(all_bars)}| Today Bar High: {bar.high} | Open: {bar.open} | Low: {bar.low} | Price: {bar.close} ")
return
invested = self.algorithm.portfolio[self.active_mes_contract].invested
current_price = self.algorithm.securities[self.mes_canonical].price
sma_200 = self._sma(all_bars, 200, 'close')
if current_price <= 0:
return
month = self.algorithm.time.month
day = self.algorithm.time.day
# Koupit po 23. červnu
if month == 6 and day >= 24 and not invested:
# Výpočet velikosti pozice pro FUTURE
portfolio_value = self.algorithm.portfolio.total_portfolio_value
risk_amount = portfolio_value * self.mes_position_exposure
contract_notion = current_price * self.algorithm.securities[self.mes_canonical].symbol_properties.contract_multiplier
if contract_notion <= 0:
return
qty_float = risk_amount / contract_notion
qty = max(1, int(qty_float)) # round up, min 1
if qty <= 0:
return
# Nákup Protective PUT/Future
self.algorithm.market_order(self.active_mes_contract, qty)
self.entry_time = self.algorithm.time
self.regime = "long"
self.algorithm.debug(f"Entry Long M2K {qty} @ ~{current_price:.1f}")
if invested and ((month == 6 and day >= 30) or month == 7):
self.algorithm.liquidate(self.active_mes_contract)
self.algorithm.debug(f"Exit Long MES @ {current_price:.2f}")
# ─── Indicators calculation ──────────────────────────────
def _sma(self, bars, period, field='close'):
if len(bars) < period:
return None
series = pd.Series([getattr(b, field) for b in bars])
sma_series = ta.sma(series, length=period)
return sma_series.iloc[-1]
def _ema(self, bars, period, field='close'):
if len(bars) < period:
return None
closes = pd.Series([b.close for b in bars])
ema_series = ta.ema(closes, length=period)
return ema_series.iloc[-1]
def _rsi(self, bars, period, field='close'):
if len(bars) < period + 1: return None
closes = pd.Series([b.close for b in bars])
rsi_series = ta.rsi(closes, length=period)
return rsi_series.iloc[-1]from AlgorithmImports import *
from datetime import timedelta, datetime
from datetime import time as dt_time
import pandas as pd
import pandas_ta as ta
class MES_Strategy:
def __init__(self, algorithm):
self.algorithm = algorithm
# ─── MNQ Future ───────────────────────────────────────────────────── Futures.Indices.MICRO_SP_500_E_MINI
self.mes_future = self.algorithm.add_future(Futures.Indices.MICRO_SP_500_E_MINI, Resolution.MINUTE, extended_market_hours=False)
self.mes_future.set_filter(30, 182)
self.mes_canonical = self.mes_future.symbol
# Managment Varibles
self.mes_position_exposure = 1
self.mes_max_days = 6
self.diagnostical_loging = True
# Empty Varibles
self.current_intraday_bar = None
self.last_date = None
self.entry_time = None
self.active_mes_contract = None
self.regime = None
self.history_bars = []
if self.diagnostical_loging:
self.algorithm.debug("MES strategy inicializována")
self.algorithm.schedule.on(self.algorithm.date_rules.every_day(),
self.algorithm.time_rules.after_market_open("SPY", 0),
self._execute_trading)
self.algorithm.schedule.on(self.algorithm.date_rules.every_day(),
self.algorithm.time_rules.before_market_close("SPY", 0),
self.execute_exit)
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day('SPY'),
self.algorithm.time_rules.every(timedelta(minutes=60)),
self._loging)
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day('SPY'),
self.algorithm.time_rules.every(timedelta(minutes=5)),
self._loging_before_trading)
def on_warm_up_finished(self):
self.algorithm.debug(f"on warm up finished in MES MODULE")
self._recover_after_restart()
self._loging()
def _recover_after_restart(self):
for holding in self.algorithm.portfolio.values():
symbol = holding.symbol
if holding.invested:
if "MES" in symbol.value:
self.active_mes_contract = symbol
holding = self.algorithm.portfolio[symbol]
self.regime = "long" if holding.quantity > 0 else "short"
key = f"entry_time_{symbol.value}"
try:
saved_time = self.algorithm.object_store.read(key)
self.entry_time = datetime.fromisoformat(saved_time)
except:
self.entry_time = self.algorithm.time
self.algorithm.debug(f"Recovered position: {symbol} | Regime: {self.regime} | Entry: {self.entry_time}| Qty: {holding.quantity}")
return
self.algorithm.debug("Not invested in MES, nothing to recover for MNQ_Module")
def on_data(self, slice: Slice):
# Aktualizace mapped futures kontraktu
if self.mes_canonical in slice.future_chains:
future_chain = slice.future_chains[self.mes_canonical]
if future_chain:
sorted_valid = sorted(future_chain, key=lambda c: c.expiry)
choosed_future = sorted_valid[0].symbol
if choosed_future != self.active_mes_contract:
if self.active_mes_contract and self.algorithm.portfolio[self.active_mes_contract].invested:
return
self.active_mes_contract = choosed_future
if not self.algorithm.is_warming_up:
self.algorithm.debug(f"New MES mapped contract: {choosed_future.value} exp: {sorted_valid[0].expiry.date()}")
# Aktualizace Daily baru
if self.mes_canonical in slice.bars:
data_slice = slice.bars[self.mes_canonical]
t = data_slice.time.time()
d = data_slice.time.date()
#if t < time(9, 30) or t > time(16, 00):
# return
if self.last_date != d:
# Save todays ended bar to history
if self.current_intraday_bar is not None:
self.history_bars.append(self.current_intraday_bar)
if len(self.history_bars) > 250:
self.history_bars.pop(0)
# Creation of new bar for today
self.current_intraday_bar = TradeBar(
data_slice.time, self.mes_canonical,
data_slice.open, data_slice.high, data_slice.low,
data_slice.close, data_slice.volume
)
self.last_date = d
else:
# Actualization of todays bar
bar = self.current_intraday_bar
bar.high = max(bar.high, data_slice.high)
bar.low = min(bar.low, data_slice.low)
bar.close = data_slice.close
bar.volume += data_slice.volume
def _loging(self):
if self.algorithm.is_warming_up or self.active_mes_contract is None:
return
t = self.algorithm.time.time()
if t < time(9, 30) or t > time(16, 00):
return
bar = self.current_intraday_bar
# ─── Calculation of indiacators from todays bar and history ───────────────────────
all_bars = self.history_bars + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
self.algorithm.debug(f"Not Enought Data History MES + Today bar for Loging | Time: {self.algorithm.time} | Bars: {len(all_bars)}| Today Bar High: {bar.high} | Open: {bar.open} | Low: {bar.low} | Price: {bar.close} ")
return
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_low = self.history_bars[-1].low if self.history_bars else None
invested = self.algorithm.portfolio[self.active_mes_contract].invested
current_price = self.algorithm.securities[self.mes_canonical].price
contract = self.active_mes_contract
ema_10 = self._ema(all_bars, 10, 'close')
sma_200 = self._sma(all_bars, 200, 'close')
self.algorithm.debug(f"MES MODULE -------- {self.algorithm.time} ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------")
self.algorithm.debug(f"Contract: {contract} | Invested: {invested} | SMA 200: {sma_200:.2f}")
self.algorithm.debug(f"Price: {current_price:.2f} | Yesterday close: {yest_close:.2f}")
self.algorithm.debug(f"Bar Close: {bar.close:.2f} | Bar High: {bar.high:.2f} | Bar Low: {bar.low:.2f} | Bar Open: {bar.open:.2f}")
def _loging_before_trading(self):
if self.algorithm.is_warming_up or self.active_mes_contract is None:
return
t = self.algorithm.time.time()
if t < time(15, 40) or t > time(16, 00):
return
bar = self.current_intraday_bar
# ─── Calculation of indiacators from todays bar and history ───────────────────────
all_bars = self.history_bars # + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
self.algorithm.debug(f"Not Enought Data History MES + Today bar for Loging | Time: {self.algorithm.time} | Bars loaded: {len(all_bars)}| Today Bar High: {bar.high} | Open: {bar.open} | Low: {bar.low} | Price: {bar.close} ")
return
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_low = self.history_bars[-1].low if self.history_bars else None
current_price = self.algorithm.securities[self.mes_canonical].price
invested = self.algorithm.portfolio[self.active_mes_contract].invested
contract = self.active_mes_contract
sma_200 = self._sma(all_bars, 200, 'close')
self.algorithm.debug(f"MES MODULE -------- {self.algorithm.time} ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------")
self.algorithm.debug(f"Contract: {contract} | Invested: {invested} | SMA 200: {sma_200:.2f}")
self.algorithm.debug(f"Price: {current_price:.2f} | Yesterday close: {yest_close:.2f}")
self.algorithm.debug(f"Bar Close: {bar.close:.2f} | Bar High: {bar.high:.2f} | Bar Low: {bar.low:.2f} | Bar Open: {bar.open:.2f}")
def _execute_trading(self):
if self.algorithm.is_warming_up or self.active_mes_contract is None:
return
invested = self.algorithm.portfolio[self.active_mes_contract].invested
current_price = self.algorithm.securities[self.mes_canonical].price
yest_low = self.history_bars[-1].low if self.history_bars else None
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_vol = self.history_bars[-1].volume if self.history_bars else None
day_before_yest_vol = self.history_bars[-2].volume if self.history_bars else None
day_bef_yest_close = self.history_bars[-2].close if len(self.history_bars) >= 2 else None
# ─── Calculation of indiacators from todays bar and history ───────────────────────
all_bars = self.history_bars # + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
return
sma_200 = self._sma(all_bars, 200, 'close')
ema_10 = self._ema(all_bars, 10, 'close')
rsi_5 = self._rsi(all_bars, 5)
if any(x is None for x in [sma_200, ema_10]):
return
if current_price <= 0:
return
# ─── ENTRY conditions ───────────────────────────────
bear_day_trading_strategy = (current_price > (yest_close * 1.01) and
rsi_5 < 50 and
current_price < sma_200 and
yest_vol < (day_before_yest_vol * 0.90))
day_trading_strategy = (current_price < yest_low and
current_price < (yest_close * 0.995) and
yest_close < ema_10 and
yest_vol < (day_before_yest_vol * 0.90))
# ─── Long Entry ─────────────────────────────────
if day_trading_strategy:
# Výpočet velikosti pozice pro FUTURE
portfolio_value = self.algorithm.portfolio.total_portfolio_value
risk_amount = portfolio_value * self.mes_position_exposure
contract_notion = current_price * self.algorithm.securities[self.mes_canonical].symbol_properties.contract_multiplier
if contract_notion <= 0:
return
qty_float = risk_amount / contract_notion
qty = max(1, int(qty_float)) # round up, min 1
if qty <= 0:
return
# Nákup Protective PUT/Future
self.algorithm.market_order(self.active_mes_contract, qty)
self.entry_time = self.algorithm.time
self.regime = "long"
self.algorithm.debug(f"Entry Long MES {qty} @ ~{current_price:.1f}")
# ─── Short Entry ─────────────────────────────────
if bear_day_trading_strategy:
# Výpočet velikosti pozice pro FUTURE
portfolio_value = self.algorithm.portfolio.total_portfolio_value
risk_amount = portfolio_value * self.mes_position_exposure
contract_notion = current_price * self.algorithm.securities[self.mes_canonical].symbol_properties.contract_multiplier
if contract_notion <= 0:
return
qty_float = risk_amount / contract_notion
qty = max(1, int(qty_float)) # round up, min 1
if qty <= 0:
return
# Nákup Protective PUT/Future
self.algorithm.market_order(self.active_mes_contract, -qty)
self.entry_time = self.algorithm.time
self.regime = "short"
self.algorithm.debug(f"Entry Short MES {qty} @ ~{current_price:.1f}")
def execute_exit (self):
if self.algorithm.is_warming_up or self.active_mes_contract is None:
return
invested = self.algorithm.portfolio[self.active_mes_contract].invested
current_price = self.algorithm.securities[self.mes_canonical].price
# ─── long exit ─────────────────────────────────────
if invested and self.regime == "long":
self.algorithm.liquidate(self.active_mes_contract)
self.algorithm.debug(f"Exit Long MES @ {current_price:.2f}")
# ─── short exit ───────────────────────────
if invested and self.regime == "short":
self.algorithm.liquidate(self.active_mes_contract)
self.algorithm.debug(f"Exit Short MES @ {current_price:.2f}")
# ─── Indicators calculation ──────────────────────────────
def _sma(self, bars, period, field='close'):
if len(bars) < period:
return None
series = pd.Series([getattr(b, field) for b in bars])
sma_series = ta.sma(series, length=period)
return sma_series.iloc[-1]
def _ema(self, bars, period, field='close'):
if len(bars) < period:
return None
closes = pd.Series([b.close for b in bars])
ema_series = ta.ema(closes, length=period)
return ema_series.iloc[-1]
def _rsi(self, bars, period, field='close'):
if len(bars) < period + 1: return None
closes = pd.Series([b.close for b in bars])
rsi_series = ta.rsi(closes, length=period)
return rsi_series.iloc[-1]from AlgorithmImports import *
from datetime import timedelta, datetime
import pandas as pd
import pandas_ta as ta
class MGC_Strategy:
def __init__(self, algorithm):
self.algorithm = algorithm
# ─── NQ Future pro obchodování ───────────────────────────────────── Futures.Metals.MICRO_GOLD
self._future = self.algorithm.add_future(Futures.Metals.GOLD, Resolution.MINUTE, extended_market_hours=True)
self._future.set_filter(90, 500)
self.future_canonical = self._future.symbol
# ─── QQQ pro signály (daily data) ───────────────────────────────────
self.gld = self.algorithm.add_equity("GLD", Resolution.DAILY)
self.gld_symbol = self.gld.symbol
self.tlt = self.algorithm.add_equity("TLT", Resolution.DAILY)
self.tlt_symbol = self.tlt.symbol
# ─── QC Indikátory pro QQQ ──────────────────────────────────────────
# Management proměnné
self._max_days = 10
self.diagnostical_loging = True
self.entry_time = None
self.active_future_contract = None
self.regime = None
self.trading = 0
self._atr_limit_order = 0.10
self._exposure = 1
self.strategy = None
self.trend_following = None
self.breakout = None
self._gld_breakout_qty = None
self.algorithm.debug("MGC strategy inicializována s GLD signály")
# ─── Schedule po uzavření QQQ trhu ──────────────────────────────────
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day("QQQ"),
self.algorithm.time_rules.after_market_close("QQQ", 0),
self._execute_trading_logic
)
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day(),
self.algorithm.time_rules.before_market_open("QQQ", 60),
self._cancel_orders
)
def on_warm_up_finished(self):
self.algorithm.debug("MGC MODULE - warm up finished")
self._recover_after_restart()
def _cancel_orders(self):
if not self.algorithm.is_warming_up:
self.algorithm.transactions.cancel_open_orders()
def _recover_after_restart(self):
for holding in self.algorithm.portfolio.values():
symbol = holding.symbol
if holding.invested and "GC" in symbol.value:
self.active_future_contract = symbol
holding_info = self.algorithm.portfolio[symbol]
self.regime = "long" if holding_info.quantity > 0 else "short"
key = f"entry_time_{symbol.value}"
try:
saved_time = self.algorithm.object_store.read(key)
self.entry_time = datetime.fromisoformat(saved_time)
except:
self.entry_time = self.algorithm.time
self.algorithm.debug(f"Recovered MGC position: {symbol} | Regime: {self.regime} | Entry: {self.entry_time} | Qty: {holding_info.quantity}")
return
self.algorithm.debug("Not invested in MGC, nothing to recover")
def on_data(self, slice: Slice):
# ─── Aktualizace mapovaného NQ kontraktu ────────────────────────────
if self.future_canonical in slice.future_chains:
future_chain = slice.future_chains[self.future_canonical]
if future_chain:
sorted_contracts = sorted(future_chain, key=lambda c: c.expiry)
new_contract = sorted_contracts[0].symbol
if new_contract != self.active_future_contract:
if self.active_future_contract and self.algorithm.portfolio[self.active_future_contract].invested:
return
self.active_future_contract = new_contract
if not self.algorithm.is_warming_up:
self.algorithm.debug(f"New NQ contract mapped: {new_contract.value} exp: {sorted_contracts[0].expiry.date()}")
def _execute_trading_logic(self):
if self.algorithm.is_warming_up or self.active_future_contract is None:
return
gld_history = self.algorithm.history(TradeBar, self.gld_symbol, 210, Resolution.DAILY)
tlt_history = self.algorithm.history(TradeBar, self.tlt_symbol, 210, Resolution.DAILY)
invested = self.algorithm.portfolio[self.active_future_contract].invested
atr_10 = self._atr(gld_history, 10)
ibs = self._ibs(gld_history)
momentum_gld = self._momentum(gld_history, 135)
momentum_tlt = self._momentum(tlt_history, 135)
contract = self.active_future_contract
three_day_high_close = self._high_breakout_close(gld_history , 4)
ten_day_high_high = self._high_breakout_high(gld_history, 11)
gld_today_close = gld_history.iloc[-1].close
gld_yesteday_high = gld_history.iloc[-2].high
if momentum_tlt is None or momentum_gld is None:
return
self.algorithm.debug(f"MGC MODULE -------- {self.algorithm.time} ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------")
self.algorithm.debug(f"Contract: {contract} | Invested: {invested} | Momentum_TLT : {momentum_tlt:.2f} | Momentum_GLD: {momentum_gld:.2f}| price: {gld_today_close:.2f} | yesterday high: {gld_yesteday_high:.2f}")
# ─── ENTRY: volatility_reversal ─────────────────────────────────────
gld_trend_following = ((momentum_tlt > 0) and (momentum_gld > 0) and not invested)
gld_breakout = ((tlt_history.iloc[-1].close > tlt_history.iloc[-2].close) and three_day_high_close == 1 and not invested and self.algorithm.time.weekday() != 4)
long_pullback = (ibs < 0.15 and
ten_day_high_high == 1 and
not invested)
if gld_trend_following:
current_price = self.algorithm.securities[self.active_future_contract].price
# Výpočet velikosti pozice
portfolio_value = self.algorithm.portfolio.total_portfolio_value
risk_amount = portfolio_value * (self._exposure / 2)
contract_multiplier = self.algorithm.securities[self.future_canonical].symbol_properties.contract_multiplier
contract_value = current_price * contract_multiplier
if contract_value <= 0:
return
qty_float = risk_amount / contract_value
qty = max(1, int(qty_float))
if qty <= 0:
return
order = self.algorithm.market_order(self.active_future_contract, int(qty))
self.entry_time = self.algorithm.time
self.regime = "long"
self.trend_following = 1
self._gld_trend_following = qty
self.algorithm.debug(f"Entry Long {qty} Current price: @ ~{current_price:.1f}")
return
# ─── EXIT: trendfollowing ──────────────────────────────
if invested:
exit_triggered = ((momentum_tlt < 0) or (momentum_gld < 0))
if exit_triggered:
self.algorithm.liquidate(self.active_future_contract)
self.algorithm.debug(f"EXIT LONG MGC @ GLD: {gld_today_close:.2f} | Včerejší high: {gld_yesteday_high:.2f}")
self.entry_time = None
self.regime = None
def _ibs(self, bar_series):
if len(bar_series) < 1: return None
if bar_series['high'].iloc[-1] == bar_series['low'].iloc[-1]:
return 0.5
ibs = (bar_series['close'].iloc[-1] - bar_series['low'].iloc[-1]) / (bar_series['high'].iloc[-1] - bar_series['low'].iloc[-1])
return ibs
def _momentum(self, bar_series, period):
if len(bar_series) < period: return None
momentum_series = ta.mom(bar_series['close'], length=period)
return momentum_series.iloc[-1]
def _atr(self, bar_series, period):
if len(bar_series) < period: return None
atr_series = ta.atr(bar_series['high'], bar_series['low'], bar_series['close'], length=period)
return atr_series.iloc[-1]
def _high_breakout_close(self, bar_series, period):
if len(bar_series) < period: return None
recent_max = bar_series['close'].iloc[-period:].max()
if bar_series['close'].iloc[-1] == recent_max:
breakout = 1
else:
breakout = 0
return breakout
def _high_breakout_high(self, bar_series, period):
if len(bar_series) < period: return None
recent_max = bar_series['high'].iloc[-period:].max()
if bar_series['high'].iloc[-1] == recent_max:
breakout = 1
else:
breakout = 0
return breakoutfrom AlgorithmImports import *
from datetime import timedelta, datetime
import pandas as pd
import numpy as np
import pandas_ta as ta
from sklearn.ensemble import RandomForestClassifier
class MNQ_Strategy:
def __init__(self, algorithm):
self.algorithm = algorithm
# ─── NQ Future pro obchodování ─────────────────────────────────────
self.nq_future = self.algorithm.add_future(Futures.Indices.NASDAQ_100_E_MINI, Resolution.MINUTE, extended_market_hours=True)
self.nq_future.set_filter(30, 182)
self.nq_canonical = self.nq_future.symbol
# ─── QQQ pro signály (daily data) ───────────────────────────────────
self.qqq = self.algorithm.add_equity("QQQ", Resolution.DAILY)
self.qqq_symbol = self.qqq.symbol
# ─── ML Regime Filter ────────────────────────────────────────────────
self.ml_model = None
self.ml_feature_cols = None
self.use_ml_filter = True # Zapnout/vypnout ML filter
self.ml_threshold = 0.3 # Pravděpodobnost unfavorable režimu
# Management proměnné
self.nq_max_days = 15
self.diagnostical_loging = True
self.entry_time = None
self.active_nq_contract = None
self.regime = None
self.trading = 0
self._atr_limit_order = 0.10
self._risk_fraction = 0.02
self.sl_price = None
self.algorithm.debug("MNQ strategy inicializována s QQQ signály + ML filter")
# ─── Schedule po uzavření QQQ trhu ──────────────────────────────────
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day("QQQ"),
self.algorithm.time_rules.after_market_close("QQQ", 0),
self._execute_trading_logic
)
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day(),
self.algorithm.time_rules.before_market_open("QQQ", 60),
self._cancel_orders
)
self.last_retrain_year = None
# Schedule retraining každý rok
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day(),
self.algorithm.time_rules.after_market_close("QQQ", 0),
self._check_and_retrain_model
)
def _check_and_retrain_model(self):
if self.last_retrain_year != self.algorithm.time.year:
self.algorithm.debug(f"Retraining ML model na datech do {self.algorithm.time}")
self._train_ml_model()
self.last_retrain_year = self.algorithm.time.year
def on_warm_up_finished(self):
self.algorithm.debug("MNQ MODULE - warm up finished, trénuji ML model...")
self._train_ml_model()
self._recover_after_restart()
def _train_ml_model(self):
"""Natrénuje ML model pro detekci unfavorable market regimes"""
try:
# Načtení historických dat (od začátku backtestu - 300 dnů warm-up)
history = self.algorithm.history(TradeBar, self.qqq_symbol, 10000, Resolution.DAILY)
if len(history) < 100:
self.algorithm.debug("ML: Nedostatek dat pro trénování, ML filter deaktivován")
self.use_ml_filter = False
return
# Konverze na DataFrame
df = pd.DataFrame({
'close': history['close'],
'high': history['high'],
'low': history['low'],
'open': history['open']
})
# Fix MultiIndex - extract just the time level
if isinstance(df.index, pd.MultiIndex):
df.index = df.index.get_level_values('time')
# ===== FEATURE ENGINEERING =====
df['returns'] = df['close'].pct_change()
# Volatility
df['volatility_20'] = df['returns'].rolling(20).std() * np.sqrt(252) * 100
df['volatility_60'] = df['returns'].rolling(60).std() * np.sqrt(252) * 100
df['vol_spike'] = (df['volatility_20'] > df['volatility_20'].rolling(60).mean() * 1.5).astype(int)
# Drawdown
df['running_max'] = df['close'].expanding().max()
df['drawdown'] = (df['close'] / df['running_max'] - 1) * 100
df['deep_drawdown'] = (df['drawdown'] < -8).astype(int)
# Momentum
df['roc_20'] = ((df['close'] / df['close'].shift(20) - 1) * 100)
df['roc_60'] = ((df['close'] / df['close'].shift(60) - 1) * 100)
df['negative_momentum'] = (df['roc_20'] < -5).astype(int)
# Moving averages
df['sma_50'] = df['close'].rolling(50).mean()
df['sma_200'] = df['close'].rolling(200).mean()
df['below_sma200'] = (df['close'] < df['sma_200']).astype(int)
df['death_cross'] = (df['sma_50'] < df['sma_200']).astype(int)
# Lower lows
df['high_20'] = df['close'].rolling(20).max()
df['low_20'] = df['close'].rolling(20).min()
df['lower_lows'] = ((df['close'] < df['high_20'].shift(20)) &
(df['low_20'] < df['low_20'].shift(20))).astype(int)
# Panic selling
df['down_day'] = (df['returns'] < 0).astype(int)
df['consecutive_down'] = df['down_day'].rolling(5).sum()
df['panic_selling'] = (df['consecutive_down'] >= 4).astype(int)
# ===== LABELING - Unfavorable periods =====
unfavorable_periods = [
(datetime(2010, 5, 1), datetime(2010, 6, 30)), # Flash Crash aftermath
(datetime(2011, 8, 1), datetime(2011, 10, 15)), # US Debt Ceiling + S&P downgrade
(datetime(2012, 5, 1), datetime(2012, 6, 30)), # Euro Debt Crisis II
(datetime(2015, 8, 17), datetime(2015, 9, 30)), # China devaluation / flash crash
(datetime(2016, 1, 1), datetime(2016, 2, 15)), # Oil crisis, China fears
(datetime(2017, 3, 21), datetime(2017, 3, 21)), # Oil crisis, China fears
(datetime(2017, 6, 8), datetime(2017, 6, 9)), # Oil crisis, China fears
(datetime(2017, 6, 26), datetime(2017, 6, 30)), # Oil crisis, China fears
(datetime(2018, 2, 1), datetime(2018, 2, 7)), # Rate hikes, trade war
(datetime(2018, 10, 1), datetime(2018, 12, 31)), # Rate hikes, trade war
(datetime(2020, 2, 15), datetime(2020, 3, 15)), # COVID crash
(datetime(2022, 1, 1), datetime(2022, 3, 11)), # Fed hikes, bear market
# (datetime(2022, 4, 5), datetime(2022, 5, 11)), # Fed hikes, bear market
# (datetime(2022, 6, 9), datetime(2022, 6, 16)), # Fed hikes, bear market
# (datetime(2022, 8, 19), datetime(2022, 10, 10)), # Fed hikes, bear market
# (datetime(2024, 7, 16), datetime(2024, 8, 2)), # Fed hikes, bear market
(datetime(2025, 2, 20), datetime(2025, 4, 8)), # Fed hikes, bear market
]
df['unfavorable'] = 0
for start, end in unfavorable_periods:
mask = (df.index >= start) & (df.index <= end)
df.loc[mask, 'unfavorable'] = 1
# Feature columns
self.ml_feature_cols = [
'volatility_20', 'volatility_60', 'vol_spike',
'drawdown', 'deep_drawdown',
'roc_20', 'roc_60', 'negative_momentum',
'below_sma200', 'death_cross', 'lower_lows',
'consecutive_down', 'panic_selling'
]
# Odstranění NaN
df_clean = df[self.ml_feature_cols + ['unfavorable']].dropna()
if len(df_clean) < 50:
self.algorithm.debug("ML: Po čištění nedostatek dat, ML filter deaktivován")
self.use_ml_filter = False
return
X = df_clean[self.ml_feature_cols]
y = df_clean['unfavorable']
# Trénování modelu
self.ml_model = RandomForestClassifier(
n_estimators=50,
max_depth=8,
min_samples_split=10,
random_state=42,
class_weight='balanced'
)
self.ml_model.fit(X, y)
# Výsledky
accuracy = self.ml_model.score(X, y)
unfavorable_pct = y.sum() / len(y) * 100
self.algorithm.debug(f"ML Model natrénován | Accuracy: {accuracy*100:.1f}% | Unfavorable periody: {unfavorable_pct:.1f}%")
except Exception as e:
self.algorithm.debug(f"ML CHYBA při trénování: {str(e)}")
self.use_ml_filter = False
def _get_ml_prediction(self, qqq_history):
"""Vrátí pravděpodobnost unfavorable režimu"""
if not self.use_ml_filter or self.ml_model is None:
return 0.0 # Pokud ML není aktivní, vždy povoleno
try:
# Vytvoření features z posledních dat
df = pd.DataFrame({
'close': qqq_history['close'],
'high': qqq_history['high'],
'low': qqq_history['low'],
'open': qqq_history['open']
})
# Fix MultiIndex - extract just the time level
if isinstance(df.index, pd.MultiIndex):
df.index = df.index.get_level_values('time')
df['returns'] = df['close'].pct_change()
df['volatility_20'] = df['returns'].rolling(20).std() * np.sqrt(252) * 100
df['volatility_60'] = df['returns'].rolling(60).std() * np.sqrt(252) * 100
df['vol_spike'] = (df['volatility_20'] > df['volatility_20'].rolling(60).mean() * 1.5).astype(int)
df['running_max'] = df['close'].expanding().max()
df['drawdown'] = (df['close'] / df['running_max'] - 1) * 100
df['deep_drawdown'] = (df['drawdown'] < -8).astype(int)
df['roc_20'] = ((df['close'] / df['close'].shift(20) - 1) * 100)
df['roc_60'] = ((df['close'] / df['close'].shift(60) - 1) * 100)
df['negative_momentum'] = (df['roc_20'] < -5).astype(int)
df['sma_50'] = df['close'].rolling(50).mean()
df['sma_200'] = df['close'].rolling(200).mean()
df['below_sma200'] = (df['close'] < df['sma_200']).astype(int)
df['death_cross'] = (df['sma_50'] < df['sma_200']).astype(int)
df['high_20'] = df['close'].rolling(20).max()
df['low_20'] = df['close'].rolling(20).min()
df['lower_lows'] = ((df['close'] < df['high_20'].shift(20)) &
(df['low_20'] < df['low_20'].shift(20))).astype(int)
df['down_day'] = (df['returns'] < 0).astype(int)
df['consecutive_down'] = df['down_day'].rolling(5).sum()
df['panic_selling'] = (df['consecutive_down'] >= 4).astype(int)
# Poslední řádek features
latest_features = df[self.ml_feature_cols].iloc[-1:].values
# Predikce
proba = self.ml_model.predict_proba(latest_features)[0]
# Handle case where model only has 1 class
if len(proba) < 2:
if len(self.ml_model.classes_) == 1:
prob_unfavorable = float(self.ml_model.classes_[0])
else:
prob_unfavorable = 0.0
else:
prob_unfavorable = proba[1]
return prob_unfavorable
except Exception as e:
self.algorithm.debug(f"ML CHYBA v predikci: {str(e)}")
return 0.0
def _cancel_orders(self):
if not self.algorithm.is_warming_up:
self.algorithm.transactions.cancel_open_orders()
def _recover_after_restart(self):
for holding in self.algorithm.portfolio.values():
symbol = holding.symbol
if holding.invested and "NQ" in symbol.value:
self.active_nq_contract = symbol
holding_info = self.algorithm.portfolio[symbol]
self.regime = "long" if holding_info.quantity > 0 else "short"
key = f"entry_time_{symbol.value}"
try:
saved_time = self.algorithm.object_store.read(key)
self.entry_time = datetime.fromisoformat(saved_time)
except:
self.entry_time = self.algorithm.time
self.algorithm.debug(f"Recovered NQ position: {symbol} | Regime: {self.regime} | Entry: {self.entry_time} | Qty: {holding_info.quantity}")
return
self.algorithm.debug("Not invested in NQ, nothing to recover")
def on_data(self, slice: Slice):
# ─── Aktualizace mapovaného NQ kontraktu ────────────────────────────
if self.nq_canonical in slice.future_chains:
future_chain = slice.future_chains[self.nq_canonical]
if future_chain:
sorted_contracts = sorted(future_chain, key=lambda c: c.expiry)
new_contract = sorted_contracts[0].symbol
if new_contract != self.active_nq_contract:
if self.active_nq_contract and self.algorithm.portfolio[self.active_nq_contract].invested:
return
self.active_nq_contract = new_contract
if not self.algorithm.is_warming_up:
self.algorithm.debug(f"New NQ contract mapped: {new_contract.value} exp: {sorted_contracts[0].expiry.date()}")
def _execute_trading_logic(self):
if self.algorithm.is_warming_up or self.active_nq_contract is None:
return
qqq_history = self.algorithm.history(TradeBar, self.qqq_symbol, 210, Resolution.DAILY)
invested = self.algorithm.portfolio[self.active_nq_contract].invested
ibs = self._ibs(qqq_history)
atr_10 = self._atr(qqq_history, 10)
momentum = self._momentum(qqq_history, 190)
contract = self.active_nq_contract
qqq_today_close = qqq_history.iloc[-1].close
qqq_yesteday_high = qqq_history.iloc[-2].high
current_price = self.algorithm.securities[self.active_nq_contract].price
if ibs is None or atr_10 is None or momentum is None:
return
# ─── ML MARKET REGIME CHECK ──────────────────────────────────────────
ml_prob_unfavorable = self._get_ml_prediction(qqq_history)
ml_unfavorable = ml_prob_unfavorable > self.ml_threshold
# Plot ML probability
self.algorithm.plot("ML Regime Filter", "Unfavorable Probability", ml_prob_unfavorable)
self.algorithm.plot("ML Regime Filter", "Threshold", self.ml_threshold)
self.algorithm.debug(f"MNQ MODULE -------- {self.algorithm.time} ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------")
self.algorithm.debug(f"Contract: {contract} | Invested: {invested} | IBS: {ibs:.2f} | ATR10: {atr_10:.2f} | momentum: {momentum:.2f}| price: {qqq_today_close:.2f} | yesterday high: {qqq_yesteday_high:.2f}")
self.algorithm.debug(f"ML FILTER: Prob Unfavorable: {ml_prob_unfavorable:.2%} | Unfavorable: {ml_unfavorable} | Filter Active: {self.use_ml_filter}")
# ─── ML FORCE EXIT: Pokud je unfavorable a máme pozici ──────────────
if invested and ml_unfavorable and self.regime == "long":
self.algorithm.liquidate(self.active_nq_contract)
self.algorithm.debug(f"ML FORCE EXIT - Unfavorable regime detected | Prob: {ml_prob_unfavorable:.2%}")
self.entry_time = None
self.regime = None
self.sl_price = None
return
# ─── ENTRY: volatility_reversal ─────────────────────────────────────
panic_selling = ((ibs < 0.1) and not invested)
# ML FILTER: Zakázat entry pokud je unfavorable
if panic_selling and not ml_unfavorable:
stoploss = current_price * 0.95
atr_pr_exposure = self._risk_fraction / (atr_10 / qqq_today_close)
# Výpočet velikosti pozice
portfolio_value = self.algorithm.portfolio.total_portfolio_value
risk_amount = portfolio_value * atr_pr_exposure
contract_multiplier = self.algorithm.securities[self.nq_canonical].symbol_properties.contract_multiplier
contract_value = current_price * contract_multiplier
if contract_value <= 0:
return
qty_float = risk_amount / contract_value
qty = max(1, int(qty_float))
if qty <= 0:
return
atr_percentage = ((self._atr_limit_order * atr_10) / qqq_today_close)
limit_price = float(current_price - (atr_percentage * current_price))
self.algorithm.market_order(self.active_nq_contract, int(qty))
self.sl_price = stoploss
self.entry_time = self.algorithm.time
self.regime = "long"
self.algorithm.debug(f"Entry Long {qty} Current price: @ ~{current_price:.1f} | ML Prob: {ml_prob_unfavorable:.2%}")
return
elif panic_selling and ml_unfavorable:
self.algorithm.debug(f"ENTRY BLOCKED by ML Filter | IBS: {ibs:.2f} | ML Prob: {ml_prob_unfavorable:.2%}")
# ─── EXIT: QQQ překonalo včerejší high ──────────────────────────────
if invested and self.regime == "long":
close_is_higher_than_yest_high = qqq_today_close > qqq_yesteday_high
ibs_abowe = ibs > 0.8
stop_loss_trigerred = self.sl_price > current_price
exit_triggered = close_is_higher_than_yest_high
# Max days exit
if self.entry_time is not None:
days_held = (self.algorithm.time - self.entry_time).days
if days_held >= self.nq_max_days:
exit_triggered = True
self.algorithm.debug(f"EXIT - max dny držení: {days_held}")
if exit_triggered:
self.algorithm.liquidate(self.active_nq_contract)
self.algorithm.debug(f"EXIT LONG NQ @ QQQ: {qqq_today_close:.2f} | Včerejší high: {qqq_yesteday_high:.2f}")
self.entry_time = None
self.regime = None
self.sl_price = None
def _ibs(self, bar_series):
if len(bar_series) < 1: return None
if bar_series['high'].iloc[-1] == bar_series['low'].iloc[-1]:
return 0.5
ibs = (bar_series['close'].iloc[-1] - bar_series['low'].iloc[-1]) / (bar_series['high'].iloc[-1] - bar_series['low'].iloc[-1])
return ibs
def _momentum(self, bar_series, period):
if len(bar_series) < period: return None
momentum_series = ta.mom(bar_series['close'], length=period)
return momentum_series.iloc[-1]
def _atr(self, bar_series, period):
if len(bar_series) < period: return None
atr_series = ta.atr(bar_series['high'], bar_series['low'], bar_series['close'], length=period)
return atr_series.iloc[-1]from AlgorithmImports import *
from datetime import timedelta, datetime
from datetime import time as dt_time
import pandas as pd
import pandas_ta as ta
class Mailhook_module:
def __init__(self, algorithm):
self.algorithm = algorithm
def mail_hook(self, symbol, quantity, price):
symbol = symbol
quantity = quantity
price = price
if self.algorithm.portfolio[symbol].quantity == 0:
direction = "EXIT"
else:
direction = "ENTRY"
message = (f"AT-01 Order mailhook | Symbol: {symbol} | Quantity: {quantity} | Price: {price}| Direction: {direction}")
self.algorithm.debug(f"AT-01 Order mailhook | Symbol: {symbol} | Quantity: {quantity} | Price: {price}| Direction: {direction}")
self.algorithm.Notify.Email("sobocik.v@gmail.com", "AT-01 Order Notification", message)
self.algorithm.Notify.Email("madera.sandra26@gmail.com", "AT-01 Order Notification", message)
self.algorithm.Notify.Email("tomas.fiser96@seznam.cz", "AT-01 Order Notification", message)
from AlgorithmImports import *
from datetime import timedelta, datetime
from datetime import time as dt_time
import pandas as pd
import pandas_ta as ta
class UB_Strategy:
def __init__(self, algorithm):
self.algorithm = algorithm
# ─── MNQ Future ─────────────────────────────────────────────────────
self.mgc_future = self.algorithm.add_future(Futures.Financials.ULTRA_US_TREASURY_BOND, Resolution.MINUTE)
self.mgc_future.set_filter(30, 182)
self.mgc_canonical = self.mgc_future.symbol
# Managment Varibles
self.mgc_position_exposure = 1
self.mgc_max_days = 10
self.diagnostical_loging = True
# Empty Varibles
self.current_intraday_bar = None
self.last_date = None
self.entry_time = None
self.active_mgc_contract = None
self.regime = None
self.history_bars = []
if self.diagnostical_loging:
self.algorithm.debug("UB strategy inicializována")
self.algorithm.schedule.on(self.algorithm.date_rules.month_end(self.mgc_canonical, 5),
self.algorithm.time_rules.before_market_close("SPY", 5),
self._execute_entry)
self.algorithm.schedule.on(self.algorithm.date_rules.month_end(self.mgc_canonical ,0),
self.algorithm.time_rules.before_market_close("SPY", 5),
self._execute_exit)
self.algorithm.schedule.on(self.algorithm.date_rules.month_start(self.mgc_canonical, 0),
self.algorithm.time_rules.after_market_open("SPY", 5),
self._execute_entry)
self.algorithm.schedule.on(self.algorithm.date_rules.month_start(self.mgc_canonical ,5),
self.algorithm.time_rules.before_market_close("SPY", 5),
self._execute_exit)
def on_warm_up_finished(self):
self.algorithm.debug(f"on warm up finished in UB MODULE")
self._recover_after_restart()
def _recover_after_restart(self):
for holding in self.algorithm.portfolio.values():
symbol = holding.symbol
if holding.invested:
if "MGC" in symbol.value:
self.active_mgc_contract = symbol
holding = self.algorithm.portfolio[symbol]
self.regime = "long" if holding.quantity > 0 else "short"
key = f"entry_time_{symbol.value}"
try:
saved_time = self.algorithm.object_store.read(key)
self.entry_time = datetime.fromisoformat(saved_time)
except:
self.entry_time = self.algorithm.time
self.algorithm.debug(f"Recovered position: {symbol} | Regime: {self.regime} | Entry: {self.entry_time}| Qty: {holding.quantity}")
return
self.algorithm.debug("Not invested in MGC, nothing to recover for MGC_Module")
def on_data(self, slice: Slice):
# Aktualizace mapped futures kontraktu
if self.mgc_canonical in slice.future_chains:
future_chain = slice.future_chains[self.mgc_canonical]
if future_chain:
sorted_valid = sorted(future_chain, key=lambda c: c.expiry)
choosed_future = sorted_valid[0].symbol
if choosed_future != self.active_mgc_contract:
if self.active_mgc_contract and self.algorithm.portfolio[self.active_mgc_contract].invested:
return
self.active_mgc_contract = choosed_future
if not self.algorithm.is_warming_up:
self.algorithm.debug(f"Nový mapped kontrakt: {choosed_future.value} exp: {sorted_valid[0].expiry.date()}")
# Aktualizace Daily baru
if self.mgc_canonical in slice.bars:
data_slice = slice.bars[self.mgc_canonical]
t = data_slice.time.time()
d = data_slice.time.date()
if t < time(9, 31) or t > time(16, 00):
return
if self.last_date != d:
# Save todays ended bar to history
if self.current_intraday_bar is not None:
self.history_bars.append(self.current_intraday_bar)
if len(self.history_bars) > 250:
self.history_bars.pop(0)
# Creation of new bar for today
self.current_intraday_bar = TradeBar(
data_slice.time, self.mgc_canonical,
data_slice.open, data_slice.high, data_slice.low,
data_slice.close, data_slice.volume
)
self.last_date = d
else:
# Actualization of todays bar
bar = self.current_intraday_bar
bar.high = max(bar.high, data_slice.high)
bar.low = min(bar.low, data_slice.low)
bar.close = data_slice.close
bar.volume += data_slice.volume
def _execute_entry(self):
if self.algorithm.is_warming_up or self.active_mgc_contract is None:
return
invested = self.algorithm.portfolio[self.active_mgc_contract].invested
current_price = self.algorithm.securities[self.mgc_canonical].price
bar = self.current_intraday_bar
ibs = self._ibs(self.current_intraday_bar)
if self.algorithm.time.day in [15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31]:
mode = 1
if self.algorithm.time.day in [1,2,3,4,5,6,7,8,9,10,11,12,13,14]:
mode = -1
if current_price <= 0:
return
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_high = self.history_bars[-1].high
day_bef_yest_close = self.history_bars[-2].close if len(self.history_bars) >= 2 else None
# ─── Long Entry ─────────────────────────────────
# Výpočet velikosti pozice pro FUTURE
portfolio_value = self.algorithm.portfolio.total_portfolio_value
risk_amount = portfolio_value * self.mgc_position_exposure
contract_notion = current_price * self.algorithm.securities[self.mgc_canonical].symbol_properties.contract_multiplier
if contract_notion <= 0:
return
qty_float = risk_amount / contract_notion
qty = max(1, int(qty_float)) # round up, min 1
if qty <= 0:
return
# Nákup Protective PUT/Future
self.algorithm.market_order(self.active_mgc_contract, (qty * mode))
self.entry_time = self.algorithm.time
self.regime = "long"
self.algorithm.debug(f"Entry Long {qty} @ ~{current_price:.1f}")
# ─── Short Entry ─────────────────────────────────
# None
def _execute_exit(self):
if self.algorithm.is_warming_up or self.active_mgc_contract is None:
return
invested = self.algorithm.portfolio[self.active_mgc_contract].invested
current_price = self.algorithm.securities[self.mgc_canonical].price
bar = self.current_intraday_bar
ibs = self._ibs(self.current_intraday_bar)
if any(x is None for x in [ibs]):
return
if current_price <= 0:
return
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_high = self.history_bars[-1].high
day_bef_yest_close = self.history_bars[-2].close if len(self.history_bars) >= 2 else None
# ─── long exit ─────────────────────────────────────
if invested and self.regime == "long":
yest_high = self.history_bars[-1].high
exit_triggered = (current_price > yest_high) or (ibs > 0.9)
# Max days exit
if self.entry_time is not None:
days_held = (self.algorithm.time - self.entry_time).days
if days_held >= self.mgc_max_days:
exit_triggered = True
self.algorithm.debug(f"Exit Long – překročena max doba držení ({days_held} dní)")
if exit_triggered:
# Zavřeme future
self.algorithm.liquidate(self.active_mgc_contract)
self.algorithm.debug(f"Exit Long @ {current_price:.2f}")
self.entry_time = None
# ─── short exit ───────────────────────────
# None
# ─── Indicators calculation ──────────────────────────────
def _ibs(self, bar):
if not bar or bar.high == bar.low: return 0.5
return (bar.close - bar.low) / (bar.high - bar.low)
def _sma(self, bars, period, field='close'):
if len(bars) < period:
return None
series = pd.Series([getattr(b, field) for b in bars])
df = pd.DataFrame({field: series})
sma_series = ta.sma(df[field], length=period)
return sma_series.iloc[-1]from AlgorithmImports import *
from datetime import timedelta, datetime
from datetime import time as dt_time
import pandas as pd
import pandas_ta as ta
class EI_Strategy:
def __init__(self, algorithm):
self.algorithm = algorithm
# ─── MNQ Future ─────────────────────────────────────────────────────
self.ei_future = self.algorithm.add_future(Futures.Indices.MICRO_SP_500_E_MINI, Resolution.MINUTE)
self.ei_future.set_filter(30, 182)
self.ei_canonical = self.ei_future.symbol
# Managment Varibles
self.ei_position_exposure = 1
self.ei_max_days = 6
self.diagnostical_loging = True
# Empty Varibles
self.current_intraday_bar = None
self.last_date = None
self.entry_time = None
self.active_ei_contract = None
self.regime = None
self.history_bars = []
if self.diagnostical_loging:
self.algorithm.debug("EI strategy inicializována")
self.algorithm.schedule.on(self.algorithm.date_rules.every_day(),
self.algorithm.time_rules.before_market_close("SPY", 5),
self._execute_trading)
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day('SPY'),
self.algorithm.time_rules.every(timedelta(minutes=60)),
self._loging)
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day('SPY'),
self.algorithm.time_rules.every(timedelta(minutes=5)),
self._loging_before_trading)
def on_warm_up_finished(self):
self.algorithm.debug(f"on warm up finished in EI MODULE")
self._recover_after_restart()
self._loging()
def _recover_after_restart(self):
for holding in self.algorithm.portfolio.values():
symbol = holding.symbol
if holding.invested:
if "FESX" in symbol.value:
self.active_ei_contract = symbol
holding = self.algorithm.portfolio[symbol]
self.regime = "long" if holding.quantity > 0 else "short"
key = f"entry_time_{symbol.value}"
try:
saved_time = self.algorithm.object_store.read(key)
self.entry_time = datetime.fromisoformat(saved_time)
except:
self.entry_time = self.algorithm.time
self.algorithm.debug(f"Recovered position: {symbol} | Regime: {self.regime} | Entry: {self.entry_time}| Qty: {holding.quantity}")
return
self.algorithm.debug("Not invested in EI, nothing to recover for MNQ_Module")
def on_data(self, slice: Slice):
# Aktualizace mapped futures kontraktu
if self.ei_canonical in slice.future_chains:
future_chain = slice.future_chains[self.ei_canonical]
if future_chain:
sorted_valid = sorted(future_chain, key=lambda c: c.expiry)
choosed_future = sorted_valid[0].symbol
if choosed_future != self.active_ei_contract:
if self.active_ei_contract and self.algorithm.portfolio[self.active_ei_contract].invested:
return
self.active_ei_contract = choosed_future
if not self.algorithm.is_warming_up:
self.algorithm.debug(f"New EI mapped contract: {choosed_future.value} exp: {sorted_valid[0].expiry.date()}")
# Aktualizace Daily baru
if self.ei_canonical in slice.bars:
data_slice = slice.bars[self.ei_canonical]
t = data_slice.time.time()
d = data_slice.time.date()
#if t < time(9, 30) or t > time(16, 00):
# return
if self.last_date != d:
# Save todays ended bar to history
if self.current_intraday_bar is not None:
self.history_bars.append(self.current_intraday_bar)
if len(self.history_bars) > 250:
self.history_bars.pop(0)
# Creation of new bar for today
self.current_intraday_bar = TradeBar(
data_slice.time, self.ei_canonical,
data_slice.open, data_slice.high, data_slice.low,
data_slice.close, data_slice.volume
)
self.last_date = d
else:
# Actualization of todays bar
bar = self.current_intraday_bar
bar.high = max(bar.high, data_slice.high)
bar.low = min(bar.low, data_slice.low)
bar.close = data_slice.close
bar.volume += data_slice.volume
def _loging(self):
if self.algorithm.is_warming_up:
return
t = self.algorithm.time.time()
if t < time(9, 30) or t > time(16, 00):
return
bar = self.current_intraday_bar
# ─── Calculation of indiacators from todays bar and history ───────────────────────
all_bars = self.history_bars + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
self.algorithm.debug(f"Not Enought Data History EI + Today bar for Loging | Time: {self.algorithm.time} | Bars: {len(all_bars)}| Today Bar High: {bar.high} | Open: {bar.open} | Low: {bar.low} | Price: {bar.close} ")
return
invested = self.algorithm.portfolio[self.active_ei_contract].invested
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_low = self.history_bars[-1].low
yest_high = self.history_bars[-1].high
day_bef_yest_close = self.history_bars[-2].close if len(self.history_bars) >= 2 else None
current_price = self.algorithm.securities[self.ei_canonical].price
contract = self.active_ei_contract
sma_200 = self._sma(all_bars, 200, 'close')
sma_190 = self._sma(all_bars, 190, 'close')
sma_10 = self._sma(all_bars, 10, 'close')
tp_10 = self._typical_price_10(all_bars)
atr_10 = self._atr(all_bars, 10)
adx_10 = self._adx(all_bars, 10)
ibs = self._ibs(self.current_intraday_bar)
self.algorithm.debug(f"EI MODULE -------- {self.algorithm.time} ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------")
self.algorithm.debug(f"Contract: {contract} | Invested: {invested} | SMA 200: {sma_200:.2f} | SMA 190: {sma_190:.2f} | SMA 10: {sma_10:.2f} | TP 10: {tp_10:.2f} | ATR 10: {atr_10:.2f} | ADX 10: {adx_10:.2f} | IBS: {ibs:.2f}")
self.algorithm.debug(f"Price: {current_price:.2f} | Yesterday close: {yest_close:.2f} | Yesterday High: {yest_high:.2f} | Yesterday Low: {yest_low:.2f} | Day before Yesterday close: {day_bef_yest_close:.2f}")
self.algorithm.debug(f"Bar Close: {bar.close:.2f} | Bar High: {bar.high:.2f} | Bar Low: {bar.low:.2f} | Bar Open: {bar.open:.2f}")
def _loging_before_trading(self):
if self.algorithm.is_warming_up:
return
t = self.algorithm.time.time()
if t < time(15, 40) or t > time(16, 00):
return
bar = self.current_intraday_bar
# ─── Calculation of indiacators from todays bar and history ───────────────────────
all_bars = self.history_bars + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
self.algorithm.debug(f"Not Enought Data History EI + Today bar for Loging | Time: {self.algorithm.time} | Bars loaded: {len(all_bars)}| Today Bar High: {bar.high} | Open: {bar.open} | Low: {bar.low} | Price: {bar.close} ")
return
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_low = self.history_bars[-1].low
yest_high = self.history_bars[-1].high
day_bef_yest_close = self.history_bars[-2].close if len(self.history_bars) >= 2 else None
current_price = self.algorithm.securities[self.ei_canonical].price
invested = self.algorithm.portfolio[self.active_ei_contract].invested
contract = self.active_ei_contract
sma_200 = self._sma(all_bars, 200, 'close')
sma_190 = self._sma(all_bars, 190, 'close')
sma_10 = self._sma(all_bars, 10, 'close')
tp_10 = self._typical_price_10(all_bars)
atr_10 = self._atr(all_bars, 10)
adx_10 = self._adx(all_bars, 10)
ibs = self._ibs(self.current_intraday_bar)
self.algorithm.debug(f"EI MODULE -------- {self.algorithm.time} ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------")
self.algorithm.debug(f"Contract: {contract} | Invested: {invested} | SMA 200: {sma_200:.2f} | SMA 190: {sma_190:.2f} | SMA 10: {sma_10:.2f} | TP 10: {tp_10:.2f} | ATR 10: {atr_10:.2f} | ADX 10: {adx_10:.2f} | IBS: {ibs:.2f}")
self.algorithm.debug(f"Price: {current_price:.2f} | Yesterday close: {yest_close:.2f} | Yesterday High: {yest_high:.2f} | Yesterday Low: {yest_low:.2f} | Day before Yesterday close: {day_bef_yest_close:.2f}")
self.algorithm.debug(f"Bar Close: {bar.close:.2f} | Bar High: {bar.high:.2f} | Bar Low: {bar.low:.2f} | Bar Open: {bar.open:.2f}")
def _execute_trading(self):
if self.algorithm.is_warming_up or self.active_ei_contract is None:
return
invested = self.algorithm.portfolio[self.active_ei_contract].invested
current_price = self.algorithm.securities[self.ei_canonical].price
# ─── Calculation of indiacators from todays bar and history ───────────────────────
all_bars = self.history_bars + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
return
sma_200 = self._sma(all_bars, 200, 'close')
sma_190 = self._sma(all_bars, 190, 'close')
sma_10 = self._sma(all_bars, 10, 'close')
tp_10 = self._typical_price_10(all_bars)
atr_10 = self._atr(all_bars, 10)
adx_10 = self._adx(all_bars, 10)
ibs = self._ibs(self.current_intraday_bar)
if any(x is None for x in [sma_190, sma_10, tp_10, atr_10, adx_10, ibs, sma_200]):
return
if current_price <= 0:
return
yest_close = self.history_bars[-1].close if self.history_bars else None
day_bef_yest_close = self.history_bars[-2].close if len(self.history_bars) >= 2 else None
# ─── ENTRY conditions ───────────────────────────────
turn_around_tuesday = (current_price < yest_close and
yest_close < day_bef_yest_close and
self.algorithm.time.weekday() == 0 and not invested)
panic_selling = (ibs < 0.1 and
(current_price > sma_190 or sma_190 is None or current_price > sma_10) and
not invested)
turn_around_tuesday_short = (current_price > yest_close and
yest_close > day_bef_yest_close and
self.algorithm.time.weekday() == 0 and
not invested and
current_price < sma_200)
volatility_edges = (current_price < tp_10 - 1.5 * atr_10 and
adx_10 > 20 and ibs < 0.6 and not invested)
# ─── Long Entry ─────────────────────────────────
if volatility_edges:
# Výpočet velikosti pozice pro FUTURE
portfolio_value = self.algorithm.portfolio.total_portfolio_value
risk_amount = portfolio_value * self.ei_position_exposure
contract_notion = current_price * self.algorithm.securities[self.ei_canonical].symbol_properties.contract_multiplier
if contract_notion <= 0:
return
qty_float = risk_amount / contract_notion
qty = max(1, int(qty_float)) # round up, min 1
if qty <= 0:
return
# Nákup Protective PUT/Future
self.algorithm.market_order(self.active_ei_contract, qty)
self.entry_time = self.algorithm.time
self.regime = "long"
self.algorithm.debug(f"Entry Long {qty} @ ~{current_price:.1f}")
# ─── Short Entry ─────────────────────────────────
"""
if turn_around_tuesday_short:
# Výpočet velikosti pozice pro FUTURE
portfolio_value = self.algorithm.portfolio.total_portfolio_value
risk_amount = portfolio_value * self.ei_position_exposure
contract_notion = current_price * self.algorithm.securities[self.ei_canonical].symbol_properties.contract_multiplier
if contract_notion <= 0:
return
qty_float = risk_amount / contract_notion
qty = max(1, int(qty_float)) # round up, min 1
if qty <= 0:
return
# Nákup Protective PUT/Future
self.algorithm.market_order(self.active_ei_contract, -qty)
self.entry_time = self.algorithm.time
self.regime = "short"
self.algorithm.debug(f"Entry Short MNQ {qty} @ ~{current_price:.1f}")
"""
# ─── long exit ─────────────────────────────────────
if invested and self.regime == "long":
yest_high = self.history_bars[-1].high
exit_triggered = current_price > yest_high
# Max days exit
if self.entry_time is not None:
days_held = (self.algorithm.time - self.entry_time).days
if days_held >= self.ei_max_days:
exit_triggered = True
self.algorithm.debug(f"Exit Long – překročena max doba držení ({days_held} dní)")
if exit_triggered:
# Zavřeme future
self.algorithm.liquidate(self.active_ei_contract)
self.algorithm.debug(f"Exit Long @ {current_price:.2f}")
self.entry_time = None
# ─── short exit ───────────────────────────
if invested and self.regime == "short":
yest_low = self.history_bars[-1].low
exit_triggered = current_price < yest_low
# Max days exit
if self.entry_time is not None:
days_held = (self.algorithm.time - self.entry_time).days
if days_held >= self.ei_max_days:
exit_triggered = True
self.algorithm.debug(f"Exit Short – překročena max doba držení ({days_held} dní)")
if exit_triggered:
# Zavřeme future
self.algorithm.liquidate(self.active_ei_contract)
self.algorithm.debug(f"Exit Short @ {current_price:.2f}")
self.entry_time = None
# ─── Indicators calculation ──────────────────────────────
def _sma(self, bars, period, field='close'):
if len(bars) < period:
return None
series = pd.Series([getattr(b, field) for b in bars])
df = pd.DataFrame({field: series})
sma_series = ta.sma(df[field], length=period)
return sma_series.iloc[-1]
def _typical_price_10(self, bars):
if len(bars) < 10: return None
s_low = self._sma(bars, 10, 'low')
s_high = self._sma(bars, 10, 'high')
s_close = self._sma(bars, 10, 'close')
return (s_low + s_high + s_close) / 3 if all([s_low, s_high, s_close]) else None
def _atr(self, bars, period):
if len(bars) < period + 1: return None
highs = pd.Series([b.high for b in bars])
lows = pd.Series([b.low for b in bars])
closes = pd.Series([b.close for b in bars])
atr_series = ta.atr(highs, lows, closes, length=period)
return atr_series.iloc[-1]
def _adx(self, bars, period):
if len(bars) < period * 2:
return None
highs = pd.Series([b.high for b in bars])
lows = pd.Series([b.low for b in bars])
closes = pd.Series([b.close for b in bars])
adx_df = ta.adx(highs, lows, closes, length=period)
return adx_df[f"ADX_{period}"].iloc[-1]
def _ibs(self, bar):
if not bar or bar.high == bar.low: return 0.5
return (bar.close - bar.low) / (bar.high - bar.low)from AlgorithmImports import *
from datetime import timedelta, datetime
from datetime import time as dt_time
import pandas as pd
import pandas_ta as ta
class UB_Strategy:
def __init__(self, algorithm):
self.algorithm = algorithm
# ─── MNQ Future ─────────────────────────────────────────────────────
self.mgc_future = self.algorithm.add_future(Futures.Financials.ULTRA_US_TREASURY_BOND, Resolution.MINUTE)
self.mgc_future.set_filter(30, 182)
self.mgc_canonical = self.mgc_future.symbol
# Managment Varibles
self.mgc_position_exposure = 1
self.mgc_max_days = 6
self.diagnostical_loging = True
# Empty Varibles
self.current_intraday_bar = None
self.last_date = None
self.entry_time = None
self.active_mgc_contract = None
self.regime = None
self.history_bars = []
if self.diagnostical_loging:
self.algorithm.debug("UB strategy inicializována")
self.algorithm.schedule.on(self.algorithm.date_rules.every_day(),
self.algorithm.time_rules.before_market_close("SPY", 5),
self._execute_trading)
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day('SPY'),
self.algorithm.time_rules.every(timedelta(minutes=60)),
self._loging)
self.algorithm.schedule.on(
self.algorithm.date_rules.every_day('SPY'),
self.algorithm.time_rules.every(timedelta(minutes=5)),
self._loging_before_trading)
def on_warm_up_finished(self):
self.algorithm.debug(f"on warm up finished in UB MODULE")
self._recover_after_restart()
self._loging()
def _recover_after_restart(self):
for holding in self.algorithm.portfolio.values():
symbol = holding.symbol
if holding.invested:
if "MGC" in symbol.value:
self.active_mgc_contract = symbol
holding = self.algorithm.portfolio[symbol]
self.regime = "long" if holding.quantity > 0 else "short"
key = f"entry_time_{symbol.value}"
try:
saved_time = self.algorithm.object_store.read(key)
self.entry_time = datetime.fromisoformat(saved_time)
except:
self.entry_time = self.algorithm.time
self.algorithm.debug(f"Recovered position: {symbol} | Regime: {self.regime} | Entry: {self.entry_time}| Qty: {holding.quantity}")
return
self.algorithm.debug("Not invested in MGC, nothing to recover for MGC_Module")
def on_data(self, slice: Slice):
# Aktualizace mapped futures kontraktu
if self.mgc_canonical in slice.future_chains:
future_chain = slice.future_chains[self.mgc_canonical]
if future_chain:
sorted_valid = sorted(future_chain, key=lambda c: c.expiry)
choosed_future = sorted_valid[0].symbol
if choosed_future != self.active_mgc_contract:
if self.active_mgc_contract and self.algorithm.portfolio[self.active_mgc_contract].invested:
return
self.active_mgc_contract = choosed_future
if not self.algorithm.is_warming_up:
self.algorithm.debug(f"Nový mapped kontrakt: {choosed_future.value} exp: {sorted_valid[0].expiry.date()}")
# Aktualizace Daily baru
if self.mgc_canonical in slice.bars:
data_slice = slice.bars[self.mgc_canonical]
t = data_slice.time.time()
d = data_slice.time.date()
if t < time(9, 31) or t > time(16, 00):
return
if self.last_date != d:
# Save todays ended bar to history
if self.current_intraday_bar is not None:
self.history_bars.append(self.current_intraday_bar)
if len(self.history_bars) > 250:
self.history_bars.pop(0)
# Creation of new bar for today
self.current_intraday_bar = TradeBar(
data_slice.time, self.mgc_canonical,
data_slice.open, data_slice.high, data_slice.low,
data_slice.close, data_slice.volume
)
self.last_date = d
else:
# Actualization of todays bar
bar = self.current_intraday_bar
bar.high = max(bar.high, data_slice.high)
bar.low = min(bar.low, data_slice.low)
bar.close = data_slice.close
bar.volume += data_slice.volume
def _loging(self):
if self.algorithm.is_warming_up:
return
t = self.algorithm.time.time()
if t < time(9, 31) or t > time(16, 00):
return
bar = self.current_intraday_bar
# ─── Calculation of indiacators from todays bar and history ───────────────────────
all_bars = self.history_bars + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
self.algorithm.debug(f"Not Enought Data History MGC + Today bar for Loging | Time: {self.algorithm.time} | Bars: {len(all_bars)}| Today Bar High: {bar.high} | Open: {bar.open} | Low: {bar.low} | Price: {bar.close} ")
return
invested = self.algorithm.portfolio[self.active_mgc_contract].invested
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_low = self.history_bars[-1].low
yest_high = self.history_bars[-1].high
day_bef_yest_close = self.history_bars[-2].close if len(self.history_bars) >= 2 else None
current_price = self.algorithm.securities[self.mgc_canonical].price
contract = self.active_mgc_contract
ibs = self._ibs(self.current_intraday_bar)
self.algorithm.debug(f"MGC MODULE -------- {self.algorithm.time} ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------")
self.algorithm.debug(f"Contract: {contract} | Invested: {invested}")
self.algorithm.debug(f"Price: {current_price:.2f} | Yesterday close: {yest_close:.2f} | Yesterday High: {yest_high:.2f} | Yesterday Low: {yest_low:.2f} | Day before Yesterday close: {day_bef_yest_close:.2f}")
self.algorithm.debug(f"Bar Close: {bar.close:.2f} | Bar High: {bar.high:.2f} | Bar Low: {bar.low:.2f} | Bar Open: {bar.open:.2f}")
def _loging_before_trading(self):
if self.algorithm.is_warming_up:
return
t = self.algorithm.time.time()
if t < time(15, 40) or t > time(16, 00):
return
bar = self.current_intraday_bar
# ─── Calculation of indiacators from todays bar and history ───────────────────────
all_bars = self.history_bars + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
self.algorithm.debug(f"Not Enought Data History MGC + Today bar for Loging | Time: {self.algorithm.time} | Bars loaded: {len(all_bars)}| Today Bar High: {bar.high} | Open: {bar.open} | Low: {bar.low} | Price: {bar.close} ")
return
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_low = self.history_bars[-1].low
yest_high = self.history_bars[-1].high
day_bef_yest_close = self.history_bars[-2].close if len(self.history_bars) >= 2 else None
current_price = self.algorithm.securities[self.mgc_canonical].price
invested = self.algorithm.portfolio[self.active_mgc_contract].invested
contract = self.active_mgc_contract
ibs = self._ibs(self.current_intraday_bar)
self.algorithm.debug(f"MGC MODULE -------- {self.algorithm.time} ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------")
self.algorithm.debug(f"Contract: {contract} | Invested: {invested}")
self.algorithm.debug(f"Price: {current_price:.2f} | Yesterday close: {yest_close:.2f} | Yesterday High: {yest_high:.2f} | Yesterday Low: {yest_low:.2f} | Day before Yesterday close: {day_bef_yest_close:.2f}")
self.algorithm.debug(f"Bar Close: {bar.close:.2f} | Bar High: {bar.high:.2f} | Bar Low: {bar.low:.2f} | Bar Open: {bar.open:.2f}")
def _execute_trading(self):
if self.algorithm.is_warming_up or self.active_mgc_contract is None:
return
invested = self.algorithm.portfolio[self.active_mgc_contract].invested
current_price = self.algorithm.securities[self.mgc_canonical].price
bar = self.current_intraday_bar
# ─── Calculation of indiacators from todays bar and history ───────────────────────
all_bars = self.history_bars + ([self.current_intraday_bar] if self.current_intraday_bar else [])
if len(all_bars) < 201:
return
sma_200 = self._sma(all_bars, 200, 'close')
sma_20 = self._sma(all_bars, 20, 'close')
adx_5 = self._adx(all_bars, 5)
ibs = self._ibs(self.current_intraday_bar)
ten_day_high = self._10_day_high(all_bars)
three_day_high_of_closes = self._3_day_high_of_close(all_bars)
if any(x is None for x in [sma_20, adx_5, ibs, sma_200, ten_day_high]):
return
if current_price <= 0:
return
yest_close = self.history_bars[-1].close if self.history_bars else None
yest_high = self.history_bars[-1].high
day_bef_yest_close = self.history_bars[-2].close if len(self.history_bars) >= 2 else None
# ─── ENTRY conditions ───────────────────────────────
panic_selling = (ibs < 0.1 and
(current_price > sma_200 or sma_200 is None or current_price > sma_20) and
not invested)
turn_around_tuesday = (current_price < yest_close and
yest_close < day_bef_yest_close and
self.algorithm.time.weekday() == 0 and not invested)
# ─── Long Entry ─────────────────────────────────
if turn_around_tuesday:
# Výpočet velikosti pozice pro FUTURE
portfolio_value = self.algorithm.portfolio.total_portfolio_value
risk_amount = portfolio_value * self.mgc_position_exposure
contract_notion = current_price * self.algorithm.securities[self.mgc_canonical].symbol_properties.contract_multiplier
if contract_notion <= 0:
return
qty_float = risk_amount / contract_notion
qty = max(1, int(qty_float)) # round up, min 1
if qty <= 0:
return
# Nákup Protective PUT/Future
#self.algorithm.market_order(self.active_mgc_contract, qty)
#self.entry_time = self.algorithm.time
#self.regime = "long"
#self.algorithm.debug(f"Entry Long {qty} @ ~{current_price:.1f}")
# ─── Short Entry ─────────────────────────────────
# None
# ─── long exit ─────────────────────────────────────
if invested and self.regime == "long":
yest_high = self.history_bars[-1].high
exit_triggered = (current_price > yest_high) or (ibs > 0.8)
# Max days exit
if self.entry_time is not None:
days_held = (self.algorithm.time - self.entry_time).days
if days_held >= self.mgc_max_days:
exit_triggered = True
self.algorithm.debug(f"Exit Long – překročena max doba držení ({days_held} dní)")
if exit_triggered:
# Zavřeme future
self.algorithm.liquidate(self.active_mgc_contract)
self.algorithm.debug(f"Exit Long @ {current_price:.2f}")
self.entry_time = None
# ─── short exit ───────────────────────────
# None
# ─── Indicators calculation ──────────────────────────────
def _sma(self, bars, period, field='close'):
if len(bars) < period:
return None
series = pd.Series([getattr(b, field) for b in bars])
df = pd.DataFrame({field: series})
sma_series = ta.sma(df[field], length=period)
return sma_series.iloc[-1]
def _10_day_high(self, bars): # high from previous 10 highs
if len(bars) < 11:
return None
highs = pd.Series([b.high for b in bars])
ten_day_high = highs.shift(1).rolling(10).max()
return ten_day_high.iloc[-1]
def _3_day_high_of_close(self, bars): # high from previous 3 Closes
if len(bars) < 3:
return None
closes = pd.Series([b.close for b in bars])
three_day_high_of_close = closes.shift(1).rolling(3).max()
return three_day_high_of_close.iloc[-1]
def _adx(self, bars, period):
if len(bars) < period * 2:
return None
highs = pd.Series([b.high for b in bars])
lows = pd.Series([b.low for b in bars])
closes = pd.Series([b.close for b in bars])
adx_df = ta.adx(highs, lows, closes, length=period)
return adx_df[f"ADX_{period}"].iloc[-1]
def _ibs(self, bar):
if not bar or bar.high == bar.low: return 0.5
return (bar.close - bar.low) / (bar.high - bar.low)from AlgorithmImports import *
from datetime import timedelta
from datetime import time as dt_time
from MNQ_module import MNQ_Strategy
from MGC_module import MGC_Strategy
from MES_module import MES_Strategy
from M2K_module import M2K_Strategy
from UB_module_seasonality import UB_Strategy
from Mailhook_module import Mailhook_module
class AT_01(QCAlgorithm):
def initialize(self):
"""Settings of Alghoritm"""
self.set_start_date(2009, 6, 1)
self.set_end_date(2025, 12, 1)
self.set_cash(1000000)
self.set_warm_up(timedelta(days=300))
self.set_brokerage_model(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.CASH)
self.set_security_initializer(lambda security: security.set_fee_model(InteractiveBrokersFeeModel()))
self.settings.seed_initial_prices = True
benchmark = self.add_equity("QQQ").symbol
self.set_benchmark(benchmark)
self.diagnostical_loging = True
self._gold_strategy = 1
self._nasdaq_100_strategy = 1
self._sp500_strategy = 0
self._rusell_2000_strategy = 0
self._bond_strategy = 1
if self._nasdaq_100_strategy == 1: self._mnq_strategy = MNQ_Strategy(self)
if self._gold_strategy == 1: self._mgc_strategy = MGC_Strategy(self)
if self._sp500_strategy == 1: self._mes_strategy = MES_Strategy(self)
if self._rusell_2000_strategy == 1: self._m2k_strategy = M2K_Strategy(self)
if self._bond_strategy == 1: self._ub_strategy = UB_Strategy(self)
self._mailhook_module = Mailhook_module(self)
self.schedule.on(
self.date_rules.every_day(),
self.time_rules.every(timedelta(minutes=60)),
self._hourly_sanity_check)
def on_data(self, slice):
if self._nasdaq_100_strategy == 1:
self._mnq_strategy.on_data(slice)
if self._gold_strategy == 1:
self._mgc_strategy.on_data(slice)
if self._sp500_strategy == 1:
self._mes_strategy.on_data(slice)
if self._rusell_2000_strategy == 1:
self._m2k_strategy.on_data(slice)
if self._bond_strategy == 1:
self._ub_strategy.on_data(slice)
def _hourly_sanity_check(self):
if not self.is_warming_up and self.diagnostical_loging:
self.debug(f"AT-01 - Main.py | Hourly Loging Sanity check - Successful | Current time: {self.time}")
def on_order_event(self, order_event):
"""Loging/Debuging of Orders"""
if order_event.status == OrderStatus.SUBMITTED:
self.debug(f"NEW ORDER: {order_event.symbol} - {order_event.fill_quantity} @ ${order_event.fill_price}")
# Save of entry time for MNQ ticker, so after restart it will recover data.
self._mailhook_module.mail_hook(order_event.symbol, order_event.fill_quantity, order_event.fill_price)
if self.live_mode:
if "NQ" in order_event.symbol.value:
key = f"entry_time_{order_event.symbol.value}"
self.object_store.save(key, str(self.time))
if "GC" in order_event.symbol.value:
key = f"entry_time_{order_event.symbol.value}"
self.object_store.save(key, str(self.time))
if "M2K" in order_event.symbol.value:
key = f"entry_time_{order_event.symbol.value}"
self.object_store.save(key, str(self.time))
if "ES" in order_event.symbol.value:
key = f"entry_time_{order_event.symbol.value}"
self.object_store.save(key, str(self.time))
if "UB" in order_event.symbol.value:
key = f"entry_time_{order_event.symbol.value}"
self.object_store.save(key, str(self.time))
elif order_event.status == OrderStatus.CANCELED:
self.debug(f"Order Canceled: {order_event.symbol}")
elif order_event.status == OrderStatus.INVALID:
self.debug(f"Order Invalid: {order_event.symbol} - {order_event.message}")
def on_brokerage_message(self, message):
"""Will Debug/Log Brokers message"""
self.log(f"BROKERAGE MESSAGE: {message}")
self.debug(f"BROKERAGE MESSAGE: {message}")
def on_warmup_finished(self) -> None:
self.debug(f"on warm up finished in MAIN CALL")
if self._nasdaq_100_strategy == 1:
self._mnq_strategy.on_warm_up_finished()
if self._gold_strategy == 1:
self._mgc_strategy.on_warm_up_finished()
if self._sp500_strategy == 1:
self._mes_strategy.on_warm_up_finished()
if self._rusell_2000_strategy == 1:
self._m2k_strategy.on_warm_up_finished()
if self._bond_strategy == 1:
self._ub_strategy.on_warm_up_finished()