| Overall Statistics |
|
Total Orders 640 Average Win 0.69% Average Loss -0.42% Compounding Annual Return 130.162% Drawdown 9.500% Expectancy 0.633 Start Equity 100000 End Equity 228938.58 Net Profit 128.939% Sharpe Ratio 3.528 Sortino Ratio 4.572 Probabilistic Sharpe Ratio 99.028% Loss Rate 38% Win Rate 62% Profit-Loss Ratio 1.65 Alpha 0.709 Beta 0.463 Annual Standard Deviation 0.218 Annual Variance 0.048 Information Ratio 2.903 Tracking Error 0.22 Treynor Ratio 1.665 Total Fees $8481.11 Estimated Strategy Capacity $0 Lowest Capacity Asset APLT X4CEBSOBHUN9 Portfolio Turnover 11.06% |
from AlgorithmImports import *
import json
import math
import pandas as pd
from datetime import timedelta
from io import StringIO
class Form4TradingAlgorithm(QCAlgorithm):
def Initialize(self):
# self.csv_file_name = "StrategyA-2012-2023.csv"
# self.csv_file_name = "StrategyA_2024.csv"
# self.csv_file_name = "StrategyA_2024_2.csv"
self.csv_file_name = "StrategyA_allInsiderTrades-2011-2023.csv"
year = 2023
self.SetStartDate(year, 1, 1) # Start backtesting period
self.SetEndDate(year, 12, 31) # End backtesting period
starting_capital = 100000
self.SetCash(starting_capital) # Initial capital for each year
self.maxSimultaneousOpenPositions = 10
self.maxInitialCapitalPerPosition = starting_capital / self.maxSimultaneousOpenPositions
self.leverage = 1 # was 3: great results
self.set_brokerage_model(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.Margin)
self.stopLossThreshold = -0.15 # -15% stop loss
self.openPositions = [] # Track open positions
self.allInsiderTrades = self.LoadAllInsiderTrades()
self.AddEquity("SPY", Resolution.Daily)
self.SetBenchmark("SPY")
# Add ticker to universe to receive pricing data
for _, insiderTrade in self.allInsiderTrades.iterrows():
ticker = insiderTrade['ticker']
self.AddEquity(ticker, Resolution.Daily)
self.Schedule.On(self.DateRules.EveryDay("SPY"), # Align with the trading calendar of SPY, excluding weekends and holidays
# SPY: Symbol of the security whose market open time is being used as a reference
# 0: This specifies how many minutes after the market opens the function should be triggered.
self.TimeRules.AfterMarketOpen("SPY", 1),
self.OnMarketOpen)
# def OnData(self, data):
def OnMarketOpen(self):
# self.Debug(f"Market open logic executed at: {self.Time}")
# Check open positions
self.CheckOpenPositions()
# Get filings for the previous trading day
todaysInsiderTrades = self.GetTodaysInsiderTrades()
if len(todaysInsiderTrades) == 0:
return
tickersToSkip = todaysInsiderTrades[todaysInsiderTrades['ticker'].apply(lambda ticker: self.Portfolio[ticker].Invested)]
if len(tickersToSkip) > 0:
tickersToSkip = tickersToSkip['ticker'].to_list()
self.Debug(f"{self.Time} - Skipping tickers: {tickersToSkip}")
# Filter insider trades to only include tickers that are not already
# included as long open position in portfolio
todaysInsiderTrades = todaysInsiderTrades[~todaysInsiderTrades['ticker'].apply(lambda ticker: self.Portfolio[ticker].Invested)]
if len(todaysInsiderTrades) == 0:
return
# Calculate available slots and capital allocation
availableSlots = self.maxSimultaneousOpenPositions - len(self.openPositions)
if availableSlots <= 0:
return
availableCash = self.Portfolio.Cash
availableBuyingPower = availableCash + (self.Portfolio.TotalPortfolioValue - availableCash) * (self.leverage - 1)
positionSize = availableBuyingPower / availableSlots
positionSize = min(positionSize, self.maxInitialCapitalPerPosition)
for _, insiderTrade in todaysInsiderTrades[:availableSlots].iterrows():
# Fetch current stock price
ticker = insiderTrade['ticker']
openPrice = self.Securities[ticker].Open
if openPrice is None or openPrice <= 0:
continue
numShares = math.floor(positionSize / openPrice)
if numShares > 0:
# self.Debug(f"Buying {ticker} at {self.Time} for ${openPrice} ({numShares} shares)")
# self.MarketOnOpenOrder(ticker, numShares)
self.MarketOrder(ticker, numShares)
self.openPositions.append({
"Symbol": ticker,
"EntryDate": self.Time,
"EntryPrice": openPrice,
"NumShares": numShares,
"StopLossPrice": openPrice * (1 + self.stopLossThreshold)
})
def CheckOpenPositions(self):
for position in self.openPositions[:]:
symbol = position["Symbol"]
stopLossPrice = position["StopLossPrice"]
entryDate = position["EntryDate"]
numShares = position["NumShares"]
currentPrice = self.Securities[symbol].Open
if currentPrice is None:
continue
# calender days
# daysHeld = (self.Time - entryDate).days
# business days
daysHeld = len(pd.date_range(start=entryDate, end=self.Time, freq='B')) - 1
if currentPrice <= stopLossPrice or daysHeld >= 5:
# self.Liquidate(symbol) # Market order at the date/time this function runs
self.MarketOrder(symbol, -1 * numShares)
self.openPositions.remove(position)
def GetTodaysInsiderTrades(self):
today = self.Time.date()
return self.allInsiderTrades[self.allInsiderTrades['date'] == pd.Timestamp(today)]
def LoadAllInsiderTrades(self):
try:
# data = self.ObjectStore.ReadBytes(self.objectStoreKey)
# if data:
# filings = json.loads(data)
# return [f for f in filings if f['Date'] == (self.Time - timedelta(days=1)).strftime('%Y-%m-%d')]
csv_data = self.ObjectStore.Read(self.csv_file_name)
# df = pd.read_csv(StringIO(csv_data), usecols=['index', 'issuerTicker', 'filedAt'], sep=",")
df = pd.read_csv(StringIO(csv_data), sep=",")
df.rename(columns={'index': 'date', 'issuerTicker': 'ticker'}, inplace=True)
df['date'] = pd.to_datetime(df['date'], format='%Y-%m-%d')
if "filedAt" in df.columns:
self.Debug("Sorting insiders by 'filedAt'")
df.sort_values(by='filedAt', inplace=True)
else:
self.Debug("Sorting insiders by 'date'")
df.sort_values(by='date', inplace=True)
self.Debug(f"Loaded {len(df):,.0f} insider trades")
return df
except Exception as e:
self.Debug(f"Error reading CSV: {e}")
return