Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Quantconnect code for shorting pumped Penny Stocks
{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"class VentralVerticalGearbox(QCAlgorithm):\n",
"\n",
" def Initialize(self):\n",
" self.SetStartDate(2019, 1, 1) # Set Start Date\n",
" self.SetEndDate(2019,2, 1) # Set End Date\n",
" self.SetCash(100000) # Set Strategy Cash\n",
" # Setup universe\n",
" self.UniverseSettings.Resolution = Resolution.Minute\n",
" self.UniverseSettings.FillForward = True\n",
" self.SetSecurityInitializer(lambda x: x.SetDataNormalizationMode(DataNormalizationMode.Raw))\n",
" self.AddUniverse(self.SelectCoarse,self.SelectFine)\n",
" \n",
" # Store yesterday's close in a dictionary for reference\n",
" self.coarseclose={} \n",
" # Only enter a trade if today's price [doubles] yesterday's close\n",
" self.targetentry=2\n",
" # Position sizing\n",
" self.numstocks = 100\n",
"\n",
" # Exit when stock rises more tha [1.5] times from the entry level\n",
" # Subject to a limit of [10] times the entry level\n",
" self.stoplevel = 1.5\n",
" self.limitlevel = 10\n",
" # Keep track of stop loss orders so we can update them\n",
" self.stops = {} \n",
" # Exclude any stocks where the data is incorrect\n",
" self.excluded = ['MDIAV','SNDE','WLL','VTL','AMMA']\n",
" \n",
" # Provide triggers to take action on corporate events\n",
" self.symbolchange = {}\n",
" self.split = {}\n",
" self.exitb4split= {}\n",
" \n",
" # Provide a trigger to exit stale trades\n",
" self.daysintrade={}\n",
" \n",
" def SelectCoarse(self, coarse):\n",
" # Penny Stock filter\n",
" myuniverse = [x for x in coarse if x.Price < 2 and x.DollarVolume < 100000]\n",
" myuniverse = [x for x in myuniverse if x.Symbol.Value not in self.excluded]\n",
" # Clear the closing price dictionary each day before re-populating it\n",
" self.coarseclose.clear() \n",
" # Save yesterday's close\n",
" for c in myuniverse:\n",
" self.coarseclose[c.Symbol] = c.Price\n",
" return [x.Symbol for x in myuniverse] # Return filtered stocks for further filtering by the SelectFine\n",
" \n",
" def SelectFine(self,fine):\n",
" fine_filter = [x.Symbol for x in fine] \n",
"\n",
" for f in fine:\n",
" # Reset these dictionaries each day\n",
" self.symbolchange[f.Symbol] = 0\n",
" self.split[f.Symbol] = 0\n",
" self.exitb4split[f.Symbol]= 0\n",
"\n",
" return fine_filter\n",
"\n",
"\n",
" def OnData(self, data):\n",
" '''OnData event is the primary entry point for your algorithm. Each new data point will be pumped in here.\n",
" Arguments:\n",
" data: Slice object keyed by symbol containing the stock data\n",
" '''\n",
" # Mark a symbol change for possible action \n",
" for kvp in data.SymbolChangedEvents:\n",
" symbol = kvp.Key\n",
" self.symbolchange[symbol] =1\n",
" \n",
" # Mark a data split so you don't get a false signal on a reverse split \n",
" # Necessary because we are using raw unadjusted data\n",
" for kvp in data.Splits:\n",
" symbol = kvp.Key\n",
" value = kvp.Value\n",
" if value.Type == 0:\n",
" self.exitb4split[symbol] =1\n",
" if value.Type == 1:\n",
" self.split[symbol] =1\n",
"\n",
" \n",
" for kvp in data:\n",
" symbol = kvp.Key\n",
" openOrders = self.Transactions.GetOpenOrders(symbol)\n",
" open = kvp.Value.Open\n",
" \n",
" # To do: cancel stale entry orders\n",
" \n",
" # Optional provision to exit positions where \n",
" # Corporate actions interfere\n",
" \n",
" if ((self.Securities[symbol].Invested and \n",
" self.Securities[symbol].HasData) and\n",
" (self.symbolchange[symbol] ==1 or \n",
" self.exitb4split[symbol] ==1)) :\n",
" #cancelledOrders = self.Transactions.CancelOpenOrders(symbol)\n",
" quantity = self.Portfolio[symbol].AbsoluteQuantity\n",
" if quantity < 0:\n",
" #exit_ticket = self.LimitOrder(symbol, -quantity,open, \"Short Cover Split or SymbolChange\")\n",
" #exit_ticket = self.MarketOrder(symbol, quantity,Tag = \"Short Cover Split or SymbolChange\")\n",
" self.symbolchange[symbol] =3 \n",
" self.exitb4split[symbol] =3\n",
" \n",
" # Optional cancellation of stale positions \n",
" if (self.Securities[symbol].Invested and \n",
" self.Securities[symbol].HasData and\n",
" #self.symbolchange[symbol] ==0 and \n",
" #self.exitb4split[symbol] ==0 and\n",
" symbol in self.daysintrade.keys()):\n",
" #self.Debug(str(self.daysintrade[symbol]))\n",
" if self.daysintrade[symbol]>=10:\n",
" #cancelledOrders = self.Transactions.CancelOpenOrders(symbol)\n",
" quantity = self.Portfolio[symbol].Quantity\n",
" if quantity < 0:\n",
" #exit_ticket = self.LimitOrder(symbol, -quantity,open, \"Exit stale trade\")\n",
" self.daysintrade[symbol]=0\n",
" \n",
" # Short entry after sharp rise from yesterday's close\n",
" # Don't enter if the rise is an illusion caused by a reverse split\n",
" if ((not self.Securities[symbol].Invested and \n",
" not openOrders) and self.Securities[symbol].HasData and\n",
" symbol in self.coarseclose.keys() and\n",
" self.split[symbol]==0 and \n",
" self.exitb4split[symbol] ==0 and \n",
" self.symbolchange[symbol] ==0 and\n",
" open >= self.coarseclose[symbol]*self.targetentry and \n",
" self.Time.hour <= 15 and self.Time.minute <=00):\n",
" \n",
" quantity = int(self.Portfolio.TotalPortfolioValue / self.numstocks / data[symbol].Close)\n",
" if quantity < 1:\n",
" continue\n",
" # Enter with limit order\n",
" enter_ticket = self.LimitOrder(symbol, -quantity,open, \"Short Entry\")\n",
" self.daysintrade[symbol]=1\n",
"\n",
" def OnOrderEvent(self, orderEvent):\n",
" if orderEvent.Status == OrderStatus.Filled: \n",
" order = self.Transactions.GetOrderById(orderEvent.OrderId)\n",
" \n",
" #Place profit taking and stop loss orders\n",
" if order.Tag == \"Short Entry\": \n",
" \n",
" quantity = orderEvent.AbsoluteFillQuantity\n",
" quarter = int(quantity / 4)\n",
" final = quantity - 3 * quarter\n",
" fill= orderEvent.FillPrice\n",
" symbol = orderEvent.Symbol\n",
" for i in range(3):\n",
" if i==0:\n",
" order1 = self.LimitOrder(symbol, quarter, fill * (1 + (i+1)*-0.1),\" 1st Profit Taking Limit Order\")\n",
" if i ==1:\n",
" order2 = self.LimitOrder(symbol, quarter, fill * (1 + (i+1)*-0.1),\"2nd Profit Taking Limit Order\")\n",
" if i==2:\n",
" order3 = self.LimitOrder(symbol, quarter, fill * (1 + (i+1)*-0.1),\"3rd Profit Taking Limit Order\") \n",
" order4 = self.LimitOrder(symbol, final, fill * 0.6,\"4th Profit Taking Limit Order\")\n",
" \n",
" # Set stop loss\n",
" self.stops[symbol] = self.StopLimitOrder(symbol, quantity, fill * self.stoplevel, fill * self.limitlevel,\"Stop Limit Order\")\n",
" \n",
" # If hit profit target, update stop order quantity\n",
" if (order.Tag == \"1st Profit Taking Limit Order\" or\n",
" order.Tag == \"2nd Profit Taking Limit Order\" or\n",
" order.Tag == \"3rd Profit Taking Limit Order\" or\n",
" order.Tag == \"4th Profit Taking Limit Order\"): \n",
" updateSettings = UpdateOrderFields()\n",
" updateSettings.Quantity = - self.Portfolio[orderEvent.Symbol].Quantity\n",
" if updateSettings.Quantity !=0:\n",
" self.stops[orderEvent.Symbol].Update(updateSettings)\n",
" else:\n",
" self.stops[orderEvent.Symbol].Cancel()\n",
" \n",
" # Increment Stale trade counter\n",
" def OnEndOfDay(self):\n",
" for kvp in self.Portfolio:\n",
" symbol = kvp.Key\n",
" holding = kvp.Value \n",
" if holding.Invested and symbol in self.daysintrade.keys():\n",
" self.daysintrade[symbol] += 1\n"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.7.9"
}
},
"nbformat": 4,
"nbformat_minor": 4
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment