| Overall Statistics |
|
Total Trades 47 Average Win 0.22% Average Loss -0.21% Compounding Annual Return 87.973% Drawdown 3.600% Expectancy -0.211 Net Profit 25.793% Sharpe Ratio 4.585 Probabilistic Sharpe Ratio 98.059% Loss Rate 61% Win Rate 39% Profit-Loss Ratio 1.03 Alpha 0.545 Beta 0.152 Annual Standard Deviation 0.122 Annual Variance 0.015 Information Ratio 2.967 Tracking Error 0.154 Treynor Ratio 3.684 Total Fees $47.00 Estimated Strategy Capacity $16000000.00 Lowest Capacity Asset CHTR UPXX4G43SIN9 |
import pandas as pd
import numpy as np
import math
from datetime import datetime, timedelta
def create_pas(np_row):
date = np_row[0]
candle_open = np_row[1]
candle_high = np_row[2]
candle_low = np_row[3]
candle_close = np_row[4]
#volume = np_row[5]
top_of_candle = max(candle_open, candle_close)
bottom_of_candle = min(candle_open, candle_close)
top_shadow = (candle_high - top_of_candle)
bottom_shadow = (bottom_of_candle - candle_low)
body_height = abs(candle_close - candle_open)
candle_height = (candle_high - candle_low)
direction = 0
if candle_close > candle_open:
direction = 1
elif candle_close < candle_open:
direction = -1
else:
if top_shadow > bottom_shadow:
direction = 1
else:
direction = -1
if top_shadow != 0:
TS = 100 * (top_shadow/candle_height)
else:
TS = 0
if bottom_shadow != 0:
BS = 100 * (bottom_shadow/candle_height)
else:
BS = 0
if body_height != 0:
BH = 100 * (body_height/candle_height)
else:
BH = 0
if candle_height != 0:
CH = candle_height
else:
CH = 0
new_candle = [date, candle_open, candle_high, candle_low, candle_close, TS, BS, BH, CH, direction]
return new_candle
def create_pas_dataset(dataset):
candles_list = dataset.values.tolist()
new_dataset = []
for x in candles_list:
new_candle = create_pas(x)
new_dataset.append(new_candle)
dataset_df = pd.DataFrame(new_dataset, columns=['Date', 'Open', 'High', 'Low', 'Close', 'TS', 'BS', 'BH', 'CH', 'Direction'])
return dataset_df
def bucketize_candle(candle, bins, labels):
TS = candle[5]
BS = candle[6]
BH = candle[7]
for x in range(5,8):
value = candle[x]
bucket = 0
for xi in bins:
if value > xi:
bucket += 1
candle[x] = bucket
candle[8] = 1
return candle
def pas_data(time, open, high, low, close):
candle = [time, open, high, low, close]
candle = create_pas(candle)
return(candle)
def get_bins(n_buckets):
boundary = int(n_buckets)
bins = np.linspace(-1, 100, boundary)
labels = range(1, boundary)
return bins, labels
def create_crypto(candle):
x = candle
crypto = str(int(x[5])) + str(int(x[6])) + str(int(x[7])) + str(int(x[8]))
crypto = int(crypto) * int(x[9])
return crypto
def get_crypto(time, open, high, low, close, n_buckets):
candle = pas_data(time, open, high, low, close)
bins, labels = get_bins(n_buckets)
candle = bucketize_candle(candle, bins, labels)
crypto = create_crypto(candle)
return cryptoimport pandas as pd
import numpy as np
import price_action_strength as PAS
def check_pct(candles_types_unique, crypto):
try:
crypto_up = candles_types_unique.loc[candles_types_unique['Original Crypto'] == crypto]
crypto_up_pct = crypto_up.iloc[0,1]
count = crypto_up.iloc[0,3]
except:
crypto_up_pct = 0
count = 0
return crypto_up_pct, count
def get_kelly_ratio(pct):
#K% = W - (1-W)/R
#K% = Kelly Percentage
#W = Winning Probability
#R = Win/Loss Ratio
KR = pct - ((1-pct)/1)
return KR
def get_trade_direction(crypto_up_pct, count, min_count, upper_limit, lower_limit):
trade_direction = 0
if (count > min_count):
if crypto_up_pct >= upper_limit or crypto_up_pct <= lower_limit:
if crypto_up_pct > 0.5:
trade_direction = 1
else:
trade_direction = -1
return trade_direction
def get_stoploss(symbol, trade_direction, quantity, open, close):
return stopimport price_action_strength as PAS
import trade
import pandas as pd
from io import StringIO
class PensiveBlueCaterpillar(QCAlgorithm):
candles_types_unique = pd.DataFrame()
ticker_list = ['AAPL', 'IBM', 'SPY', 'GOOG', 'FB', 'TSLA', 'NFLX', 'AMZN', 'NVDA', 'AMD', 'TWTR', 'TMUS', 'WMT', 'MS', 'BAC', 'BLK', 'BP', 'F', 'GE', 'MMM', 'XOM', 'CVX', 'BABA', 'V', 'JPM', 'UNH', 'PYPL', 'DIS', 'ADBE', 'NKE', 'PFE', 'ORCL', 'XOM', 'KO', 'CRM', 'VZ', 'NFLX', 'INTC', 'PEP', 'WFC', 'ACN', 'T', 'CVX', 'MS', 'COST', 'SHOP', 'TMUS', 'MCD', 'AZN', 'SAP', 'UPS', 'QCOM', 'PM', 'SE', 'MRNA', 'C', 'BMY', 'UL', 'INTU', 'CHTR']
#ticker_list = ['AAPL', 'SPY']
debug_list = []
n_buckets = 50
min_count = 20
upper_limit = 0.70
lower_limit = 0.30
def Initialize(self):
if self.n_buckets == 10:
self.csv = self.Download("https://quantclone.s3.us-east-2.amazonaws.com/Candle+Types+-+10-1.csv")
elif self.n_buckets == 25:
self.csv = self.Download("https://quantclone.s3.us-east-2.amazonaws.com/Candle+Types+-+25-1.csv")
elif self.n_buckets == 50:
self.csv = self.Download("https://quantclone.s3.us-east-2.amazonaws.com/Candle+Types+-+50-1.csv")
else:
self.csv = self.Download("https://quantclone.s3.us-east-2.amazonaws.com/Candle+Types+-+100-1.csv")
self.candles_types_unique = pd.read_csv(StringIO(self.csv))
self.SetStartDate(2021, 8, 11)
self.SetEndDate(2021, 12, 22)
self.SetCash(10000)
self.add_tickers(self.ticker_list)
#self.min_count = self.GetParameter('min_count')
#self.upper_limit = self.GetParameter('upper_limit')
#self.lower_limit = self.GetParameter('lower_limit')
def OnData(self, data):
for x in self.ticker_list:
if data.Bars.ContainsKey(x):
time = data[x].Time
open = data[x].Open
high = data[x].High
low = data[x].Low
close = data[x].Close
def makeTrade(self, time, open, high, low, close, x):
crypto = PAS.get_crypto(time, open, high, low, close, self.n_buckets)
crypto_up_pct, count = trade.check_pct(self.candles_types_unique, crypto)
trade_direction = trade.get_trade_direction(crypto_up_pct, count, self.min_count, self.upper_limit, self.lower_limit)
risk_pct = trade.get_kelly_ratio(crypto_up_pct)
quantity = self.CalculateOrderQuantity(x, 0.25)
if trade_direction == 1:
qty = quantity
self.Debug('Long: ' + str(x) + ' Shares: ' + str(qty) + ' Time: ' + str(time))
self.MarketOrder(x, qty)
#marketOnOpenOrderTicket = self.MarketOnOpenOrder(x, qty)
marketCloseOrderTicket = self.MarketOnCloseOrder(x, qty * -1)
stopMarketTicket = self.StopMarketOrder(x, qty, close * .9)
tpTicket = self.StopMarketOrder(x, qty, close * 1.05)
if trade_direction == -1:
qty = (quantity) * -1
self.Debug('Short: ' + str(x) + ' Shares: ' + str(qty) + ' Time: ' + str(time))
self.MarketOrder(x, qty)
#marketOnOpenOrderTicket = self.MarketOnOpenOrder(x, qty)
marketCloseOrderTicket = self.MarketOnCloseOrder(x, qty * -1)
stopMarketTicket = self.StopMarketOrder(x, qty, close * 1.05)
tpTicket = self.StopMarketOrder(x, qty, close * .95)
def DailyBarHandler(self, Tradebar):
time = Tradebar.EndTime
open = Tradebar.Open
high = Tradebar.High
low = Tradebar.Low
close = Tradebar.Close
self.makeTrade(time, open, high, low, close, Tradebar.Symbol)
def OnEndOfAlgorithm(self):
self.Debug("is working the end of algo function")
self.Debug(self.debug_list)
self.Debug('ferp')
def add_tickers(self, ticker_list):
for x in ticker_list:
self.AddEquity(x, Resolution.Hour)
self.Consolidate(x, Resolution.Daily, self.DailyBarHandler)
return
#if not self.Portfolio.Invested:
#if data.Bars.ContainsKey(x):