| Overall Statistics |
|
Total Trades 8684 Average Win 0.21% Average Loss -0.06% Compounding Annual Return 25.229% Drawdown 20.400% Expectancy 1.301 Net Profit 1943.111% Sharpe Ratio 1.302 Probabilistic Sharpe Ratio 77.354% Loss Rate 50% Win Rate 50% Profit-Loss Ratio 3.63 Alpha 0.211 Beta 0.096 Annual Standard Deviation 0.169 Annual Variance 0.029 Information Ratio 0.494 Tracking Error 0.238 Treynor Ratio 2.3 Total Fees $9066.20 Estimated Strategy Capacity $7800000.00 Lowest Capacity Asset INTC R735QTJ8XC9X |
# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
# Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from clr import AddReference
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Algorithm.Framework")
AddReference("QuantConnect.Indicators")
from QuantConnect import *
from QuantConnect.Indicators import *
from QuantConnect.Algorithm import *
from QuantConnect.Algorithm.Framework import *
from QuantConnect.Algorithm.Framework.Alphas import *
import numpy as np
import pandas as pd
import math
class this_alpha_module(AlphaModel):
'''Alpha model that uses an EMA cross to create insights'''
def __init__(self, max_holding_period=250, resolution = Resolution.Daily, param_dict=None, data_dict=None, when_trigger=None, assets_dict=None, assets_history_df=None, quantity_dict=None):
self.param_dict=param_dict
self.data=data_dict
#self.assets_dict=assets_dict
self.history=assets_history_df
self.quantity=quantity_dict
self.predictionInterval = Time.Multiply(Extensions.ToTimeSpan(resolution), max_holding_period)
self.very_first_time=True
self.consol_counting={}
resolutionString = Extensions.GetEnumString(resolution, Resolution)
self.Name = '{}({}_{})'.format(self.__class__.__name__, resolutionString, when_trigger)
self.latest_rebalance_date=None
self.when_trigger=when_trigger
self.tlt_symbol = Symbol.Create('TLT', SecurityType.Equity, Market.USA)
self.assets_dict={}
# Market and list of signals based on ETFs
def Update(self, algorithm, data):
'''Updates this alpha model with the latest data from the algorithm.
This is called each time the algorithm receives data for subscribed securities
Args:
algorithm: The algorithm instance
data: The new data available
Returns:
The new insights generated'''
if algorithm.Time.date()==self.latest_rebalance_date:
return []
if algorithm.Time.hour>=10:
self.insights=[]
self.rebalance_when_out_of_the_market(algorithm)
self.latest_rebalance_date=algorithm.Time.date()
algorithm.Log('done rebalancing for insight, len of insight is {}'.format(len(self.insights)))
return self.insights
def OnSecuritiesChanged(self, algorithm, changes):
'''Event fired each time the we add/remove securities from the data feed
Args:
algorithm: The algorithm instance that experienced the change in securities
changes: The security additions and removals from the algorithm'''
if self.very_first_time==True:
self.very_first_time=False
self.assets_dict['MRKT'] = algorithm.AddEquity('SPY', Resolution.Daily).Symbol # market
self.assets_dict['GOLD'] = algorithm.AddEquity('GLD', Resolution.Daily).Symbol # gold
self.assets_dict['SLVA'] = algorithm.AddEquity('SLV', Resolution.Daily).Symbol # vs silver
self.assets_dict['UTIL'] = algorithm.AddEquity('XLU', Resolution.Daily).Symbol # utilities
self.assets_dict['INDU'] = algorithm.AddEquity('XLI', Resolution.Daily).Symbol # vs industrials
self.assets_dict['METL'] = algorithm.AddEquity('DBB', Resolution.Daily).Symbol # input prices (metals)
self.assets_dict['USDX'] = algorithm.AddEquity('UUP', Resolution.Daily).Symbol # safe haven (USD)
self.FORPAIRS = [self.assets_dict['GOLD'], self.assets_dict['SLVA'], self.assets_dict['UTIL'], self.assets_dict['INDU'], self.assets_dict['METL'], self.assets_dict['USDX']]
# Setup daily consolidation
symbols = [self.assets_dict['MRKT']] + self.FORPAIRS
for symbol in symbols:
self.consol_counting[symbol]=0
self.consolidator = TradeBarConsolidator(timedelta(days=1))
self.consolidator.DataConsolidated += self.consolidation_handler
algorithm.SubscriptionManager.AddConsolidator(symbol, self.consolidator)
# Warm up history
self.history = algorithm.History(symbols, self.param_dict['lookback'], Resolution.Daily)
if self.history.empty or 'close' not in self.history.columns:
return
self.history = self.history['close'].unstack(level=0).dropna()
#algorithm.Log('very first index and last index of history dict is {} and {}'.format(self.history.index[0],self.history.index[-1]))
#algorithm.Log('securities changes, in alpha model, {} added'.format(len(changes.AddedSecurities)))
addedSymbols = []
for security in changes.AddedSecurities:
addedSymbols.append(security.Symbol)
if security.Symbol not in self.data:
self.data[security.Symbol] = SymbolData(security.Symbol, self.param_dict['formation_days'], algorithm)
if len(addedSymbols) > 0:
history = algorithm.History(addedSymbols, 1 + self.param_dict['formation_days'], Resolution.Daily).loc[addedSymbols]
for symbol in addedSymbols:
try:
self.data[symbol].Warmup(history.loc[symbol])
except:
self.Debug(str(symbol))
continue
def derive_vola_waitdays(self):
volatility = 0.6 * np.log1p(self.history[[self.assets_dict['MRKT']]].pct_change()).std() * np.sqrt(252)
wait_days = int(volatility * self.param_dict['WAITD_CONSTANT'])
returns_lookback = int((1.0 - volatility) * self.param_dict['WAITD_CONSTANT'])
return wait_days, returns_lookback
def rebalance_when_out_of_the_market(self, algorithm):
self.param_dict['wait_days'], returns_lookback = self.derive_vola_waitdays()
## Check for Bears
algorithm.Log('history first index is {}'.format(self.history.index[0]))
for key in self.consol_counting.keys():
algorithm.Log('consol updated {} so far for {}'.format(key, self.consol_counting[key]))
returns = self.history.pct_change(returns_lookback).iloc[-1]
silver_returns = returns[self.assets_dict['SLVA']]
gold_returns = returns[self.assets_dict['GOLD']]
industrials_returns = returns[self.assets_dict['INDU']]
utilities_returns = returns[self.assets_dict['UTIL']]
metals_returns = returns[self.assets_dict['METL']]
dollar_returns = returns[self.assets_dict['USDX']]
#algorithm.Log('g last index is {}'.format(self.history.index[-1]))
#algorithm.Log('s is {}'.format(silver_returns))
#algorithm.Log('u is {}'.format(utilities_returns))
#algorithm.Log('i is {}'.format(industrials_returns))
#algorithm.Log('m is {}'.format(metals_returns))
#algorithm.Log('d is {}'.format(dollar_returns))
self.param_dict['DISTILLED_BEAR'] = (((gold_returns > silver_returns) and
(utilities_returns > industrials_returns)) and
(metals_returns < dollar_returns)
)
#algorithm.Log('dist is {}'.format(self.param_dict['DISTILLED_BEAR']))
# Determine whether 'in' or 'out' of the market
if self.param_dict['DISTILLED_BEAR']:
self.param_dict['BE_IN'] = False
self.param_dict['OUTDAY'] = self.param_dict['DCOUNT']
if self.quantity[self.tlt_symbol] == 0:
for symbol in self.quantity.copy().keys():
if symbol == self.tlt_symbol:
continue
#self.Order(symbol, - self.quantity[symbol])
self.insights.append(Insight.Price(symbol, self.predictionInterval, InsightDirection.Flat))
#algorithm.Debug([str(algorithm.Time), str(symbol), str(-self.quantity[symbol])])
del self.quantity[symbol]
quantity = algorithm.Portfolio.TotalPortfolioValue / algorithm.Securities[self.tlt_symbol].Close
self.quantity[self.tlt_symbol] = math.floor(quantity)
#self.Order(self.BND1, self.quantity[self.BND1])
self.insights.append(Insight.Price(self.tlt_symbol, self.predictionInterval, InsightDirection.Up))
#algorithm.Debug([str(algorithm.Time), str(self.tlt_symbol), str(self.quantity[self.tlt_symbol])])
algorithm.Log('flipping to tlt today')
if (self.param_dict['DCOUNT'] >= self.param_dict['OUTDAY'] + self.param_dict['wait_days']):
self.param_dict['BE_IN'] = True
#algorithm.Log('distilled is {}'.format(self.param_dict['DISTILLED_BEAR']))
#algorithm.Log('be_in is {}'.format(self.param_dict['BE_IN']))
#algorithm.Log('dcount is {}'.format(self.param_dict['DCOUNT']))
#algorithm.Log('reb_count is {}'.format(self.param_dict['reb_count']))
# Update stock ranking/holdings, when swithing from 'out' to 'in' plus every X days when 'in' (set rebalance frequency)
if (self.param_dict['BE_IN'] and not self.param_dict['BE_IN_PRIOR']) or (self.param_dict['BE_IN'] and (self.param_dict['DCOUNT']==self.param_dict['reb_count'])):
self.rebalance(algorithm)
self.param_dict['BE_IN_PRIOR'] = self.param_dict['BE_IN']
self.param_dict['DCOUNT'] += 1
def rebalance(self, algorithm):
algorithm.Log('triggering rebalance, ie buying stocks and selling tlt')
if self.param_dict['symbols'] is None: return
chosen_df = self.calc_return(self.param_dict['symbols'])
chosen_df = chosen_df.iloc[:self.param_dict['num_stocks']]
if self.quantity[self.tlt_symbol] > 0:
#self.Order(self.BND1, - self.quantity[self.BND1])
self.insights.append(Insight.Price(self.tlt_symbol, self.predictionInterval, InsightDirection.Flat))
#algorithm.Debug([str(algorithm.Time), str(self.tlt_symbol), str(-self.quantity[self.tlt_symbol])])
self.quantity[self.tlt_symbol] = 0
weight = 1 / self.param_dict['num_stocks']
for symbol in self.quantity.copy().keys():
if symbol == self.tlt_symbol:
continue
if not algorithm.CurrentSlice.ContainsKey(symbol) or algorithm.CurrentSlice[symbol] is None:
continue
if symbol not in chosen_df.index:
#self.Order(symbol, - self.quantity[symbol])
self.insights.append(Insight.Price(symbol, self.predictionInterval, InsightDirection.Flat))
#algorithm.Debug([str(algorithm.Time), str(symbol), str(-self.quantity[symbol])])
del self.quantity[symbol]
else:
quantity = algorithm.Portfolio.TotalPortfolioValue * weight / algorithm.Securities[symbol].Close
if math.floor(quantity) != self.quantity[symbol]:
#self.Order(symbol, math.floor(quantity) - self.quantity[symbol])
self.insights.append(Insight.Price(symbol, self.predictionInterval, InsightDirection.Up))
#algorithm.Debug([str(algorithm.Time), str(symbol), str(math.floor(quantity) -self.quantity[symbol])])
self.quantity[symbol] = math.floor(quantity)
algorithm.Log('chosen_df in alpha model contains {}'.format(chosen_df.index))
for symbol in chosen_df.index:
if not algorithm.CurrentSlice.ContainsKey(symbol) or algorithm.CurrentSlice[symbol] is None:
continue
if symbol not in self.quantity.keys():
quantity = algorithm.Portfolio.TotalPortfolioValue * weight / algorithm.Securities[symbol].Close
self.quantity[symbol] = math.floor(quantity)
#self.Order(symbol, self.quantity[symbol])
self.insights.append(Insight.Price(symbol, self.predictionInterval, InsightDirection.Up))
#algorithm.Debug([str(algorithm.Time), str(symbol), str(self.quantity[symbol])])
algorithm.Log('flipping to stocks')
def calc_return(self, stocks):
#ready = [self.data[symbol] for symbol in stocks if self.data[symbol].Roc.IsReady]
#sorted_by_roc = sorted(ready, key=lambda x: x.Roc.Current.Value, reverse = not self.lowmom)
#return [symbol_data.Symbol for symbol_data in sorted_by_roc[:self.num_stocks] ]
ret = {}
for symbol in stocks:
try:
ret[symbol] = self.data[symbol].Roc.Current.Value
except:
# self.Debug(str(symbol))
continue
df_ret = pd.DataFrame.from_dict(ret, orient='index')
df_ret.columns = ['return']
sort_return = df_ret.sort_values(by = ['return'], ascending = self.param_dict['lowmom'])
return sort_return
def consolidation_handler(self, sender, consolidated):
self.history.loc[consolidated.EndTime, consolidated.Symbol] = consolidated.Close
self.history = self.history.iloc[-self.param_dict['lookback']:]
self.consol_counting[consolidated.Symbol]+=1
class SymbolData(object):
def __init__(self, symbol, roc, algorithm):
self.Symbol = symbol
self.Roc = RateOfChange(roc)
self.consolidator = algorithm.ResolveConsolidator(symbol, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.Roc, self.consolidator)
def Warmup(self, history):
for index, row in history.iterrows():
self.Roc.Update(index, row['close'])"""
SEL(stock selection part)
Based on the 'Momentum Strategy with Market Cap and EV/EBITDA' strategy introduced by Jing Wu, 6 Feb 2018
adapted and recoded by Jack Simonson, Goldie Yalamanchi, Vladimir, Peter Guenther, Leandro Maia, Simone Pantaleoni, Mirko Vari(Strongs)
https://www.quantconnect.com/forum/discussion/3377/momentum-strategy-with-market-cap-and-ev-ebitda/p1
https://www.quantconnect.com/forum/discussion/9678/quality-companies-in-an-uptrend/p1
https://www.quantconnect.com/forum/discussion/9632/amazing-returns-superior-stock-selection-strategy-superior-in-amp-out-strategy/p1
I/O(in & out part)
The Distilled Bear in & out algo
based on Dan Whitnable's 22 Oct 2020 algo on Quantopian.
Dan's original notes:
"This is based on Peter Guenther great “In & Out” algo.
Included Tentor Testivis recommendation to use volatility adaptive calculation of WAIT_DAYS and RET.
Included Vladimir's ideas to eliminate fixed constants
Help from Thomas Chang"
https://www.quantopian.com/posts/new-strategy-in-and-out
https://www.quantconnect.com/forum/discussion/9597/the-in-amp-out-strategy-continued-from-quantopian/
"""
from QuantConnect.Data.UniverseSelection import *
import math
import numpy as np
import pandas as pd
import scipy as sp
from This_Universe_Module import this_universe_module
from This_Alpha_Module import this_alpha_module
class EarningsFactorWithMomentum_InOut(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2008, 1, 1) #Set Start Date
#self.SetEndDate(2008, 3, 1) #Set Start Date
self.cap = 10000
self.SetCash(self.cap)
self.averages = {}
res = Resolution.Hour
self.param_dict={}
# Holdings
### 'Out' holdings and weights
self.BND1 = self.AddEquity('TLT', res).Symbol #TLT; TMF for 3xlev
self.quantity = {self.BND1: 0}
##### In & Out parameters #####
# Feed-in constants
self.param_dict['INI_WAIT_DAYS'] = 15 # out for 3 trading weeks
self.param_dict['wait_days'] = self.param_dict['INI_WAIT_DAYS']
# Specific variables
self.param_dict['DISTILLED_BEAR'] = 1#999
self.param_dict['BE_IN'] = 1#999
self.param_dict['BE_IN_PRIOR'] = 0
self.param_dict['VOLA_LOOKBACK'] = 126
self.param_dict['WAITD_CONSTANT'] = 85
self.param_dict['DCOUNT'] = 0 # count of total days since start
self.param_dict['OUTDAY'] = (-self.param_dict['INI_WAIT_DAYS']+1) # dcount when self.be_in=0, initial setting ensures trading right away
# set a warm-up period to initialize the indicator
# self.SetWarmUp(timedelta(350))
# no warmup, will use History instead because some of the day_count items are counting during warmup
##### Momentum & fundamentals strategy parameters #####
self.UniverseSettings.Resolution = res
self.param_dict['num_coarse'] = 100#200
self.param_dict['num_screener'] = 20#100 # changed from 15
self.param_dict['num_stocks'] = 10 # lowered from 10
self.param_dict['formation_days'] = 126
self.param_dict['lowmom'] = False
self.data = {}
self.param_dict['setrebalancefreq'] = 60 # X days, update universe and momentum calculation
self.param_dict['updatefinefilter'] = 0
self.param_dict['symbols'] = None
self.param_dict['reb_count'] = 0
#self.Schedule.On(
# self.DateRules.EveryDay(),
# self.TimeRules.AfterMarketOpen('SPY', 30), # reduced time
# self.rebalance_when_out_of_the_market)
#self.Schedule.On(
# self.DateRules.EveryDay(),
# self.TimeRules.BeforeMarketClose('SPY', 0),
# self.record_vars)
# Benchmark = record SPY
self.spy = []
# Warm up history
self.param_dict['lookback'] = 126
### do the framework magic
self.AddUniverseSelection(this_universe_module(param_dict=self.param_dict, averages_dict=self.averages))
self.AddAlpha(this_alpha_module(param_dict=self.param_dict, data_dict=self.data, when_trigger=30, assets_dict=None, assets_history_df=None, quantity_dict=self.quantity)) # the original schedule event now sits in the alpha_model
self.SetPortfolioConstruction( EqualWeightingPortfolioConstructionModel())
self.SetExecution( ImmediateExecutionModel())
self.AddRiskManagement( NullRiskManagementModel())
# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
# Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm.Framework")
from QuantConnect.Data.UniverseSelection import *
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel
from itertools import groupby
from math import ceil
import numpy as np
class this_universe_module(FundamentalUniverseSelectionModel):
def __init__(self, filterFineData = True, universeSettings = None, securityInitializer = None, param_dict=None, averages_dict=None):
'''Initializes a new default instance of the QC500UniverseSelectionModel'''
super().__init__(filterFineData, universeSettings, securityInitializer)
self.param_dict=param_dict
self.averages=averages_dict
def SelectCoarse(self, algorithm, coarse):
# Update at the beginning (by setting self.OUTDAY = -self.INI_WAIT_DAYS), every X days (rebalance frequency), and one day before waitdays are up
algorithm.Log('doing universe selection...current DCOUNT is {}'.format(self.param_dict['DCOUNT']))
if not (((self.param_dict['DCOUNT']-self.param_dict['reb_count'])==self.param_dict['setrebalancefreq']) or (self.param_dict['DCOUNT'] == self.param_dict['OUTDAY'] + self.param_dict['wait_days'] - 1)):
self.param_dict['updatefinefilter'] = 0
return Universe.Unchanged
self.param_dict['updatefinefilter'] = 1
# drop stocks which have no fundamental data or have too low prices
selected = [x for x in coarse if (x.HasFundamentalData) and (float(x.Price) > 5)]
# We are going to use a dictionary to refer the object that will keep the moving averages
for cf in selected:
symbol = cf.Symbol
if cf.Symbol not in self.averages:
self.averages[cf.Symbol] = SymbolDataVolume(cf.Symbol, 21, 5, 504)
# Updates the SymbolData object with current EOD price
avg = self.averages[cf.Symbol]
avg.update(cf.EndTime, cf.AdjustedPrice, cf.DollarVolume)
# Filter the values of the dict: we only want up-trending securities
#values = list(filter(lambda sd: sd.smaw.Current.Value > 0, self.averages.values()))
values = list(filter(lambda sd: sd.smaw.Current.Value > 0, self.averages.values()))
values.sort(key=lambda x: (x.smaw.Current.Value), reverse=True)
#values.sort(key=lambda x: (x.volume_ratiol), reverse=True)
# we need to return only the symbol objects
coarse_output=[ x.symbol for x in values[:self.param_dict['num_coarse']] ]
algorithm.Log('coarse output len is {}'.format(len(coarse_output)))
return coarse_output
def SelectFine(self, algorithm, fine):
if self.param_dict['updatefinefilter'] == 0:
return Universe.Unchanged
filtered_fundamental = [x for x in fine if
x.FinancialStatements.IncomeStatement.NetIncome.TwelveMonths and
x.FinancialStatements.CashFlowStatement.CashFlowFromContinuingOperatingActivities.TwelveMonths and
x.OperationRatios.ROA.ThreeMonths and x.OperationRatios.ROA.OneYear and
x.FinancialStatements.BalanceSheet.ShareIssued.ThreeMonths and x.FinancialStatements.BalanceSheet.ShareIssued.TwelveMonths and
x.OperationRatios.GrossMargin.ThreeMonths and x.OperationRatios.GrossMargin.OneYear and
x.OperationRatios.LongTermDebtEquityRatio.ThreeMonths and x.OperationRatios.LongTermDebtEquityRatio.OneYear and
x.OperationRatios.CurrentRatio.ThreeMonths and x.OperationRatios.CurrentRatio.OneYear and
x.OperationRatios.AssetsTurnover.ThreeMonths and x.OperationRatios.AssetsTurnover.OneYear and x.ValuationRatios.NormalizedPERatio]
sortedByfactor1 = [x for x in fine if FScore(x.FinancialStatements.IncomeStatement.NetIncome.TwelveMonths,
x.FinancialStatements.CashFlowStatement.CashFlowFromContinuingOperatingActivities.TwelveMonths,
x.OperationRatios.ROA.ThreeMonths, x.OperationRatios.ROA.OneYear,
x.FinancialStatements.BalanceSheet.ShareIssued.ThreeMonths, x.FinancialStatements.BalanceSheet.ShareIssued.TwelveMonths,
x.OperationRatios.GrossMargin.ThreeMonths, x.OperationRatios.GrossMargin.OneYear,
x.OperationRatios.LongTermDebtEquityRatio.ThreeMonths, x.OperationRatios.LongTermDebtEquityRatio.OneYear,
x.OperationRatios.CurrentRatio.ThreeMonths, x.OperationRatios.CurrentRatio.OneYear,
x.OperationRatios.AssetsTurnover.ThreeMonths, x.OperationRatios.AssetsTurnover.OneYear).ObjectiveScore() > 6]
top = sorted(sortedByfactor1, key = lambda x: x.ValuationRatios.NormalizedPERatio, reverse=False)[:self.param_dict['num_screener']]
self.param_dict['symbols'] = [x.Symbol for x in top]
self.param_dict['updatefinefilter'] = 0
self.param_dict['reb_count'] = self.param_dict['DCOUNT']
algorithm.Log('done universe selection, total {}'.format(len(self.param_dict['symbols'])))
return self.param_dict['symbols']
class SymbolDataVolume(object):
def __init__(self, symbol, period, periodw, periodlt):
self.symbol = symbol
#self.tolerance = 1.01
self.tolerance = 0.95
self.fast = ExponentialMovingAverage(10)
self.slow = ExponentialMovingAverage(21)
self.is_uptrend = False
self.scale = 0
self.volume = 0
self.volume_ratio = 0
self.volume_ratiow = 0
self.volume_ratiol = 0
self.sma = SimpleMovingAverage(period)
self.smaw = SimpleMovingAverage(periodw)
self.smalt = SimpleMovingAverage(periodlt)
def update(self, time, value, volume):
self.volume = volume
if self.smaw.Update(time, volume):
# get ratio of this volume bar vs previous 10 before it.
if self.smaw.Current.Value != 0:
self.volume_ratiow = volume / self.smaw.Current.Value
if self.sma.Update(time, volume):
# get ratio of this volume bar vs previous 10 before it.
if self.sma.Current.Value != 0:
self.volume_ratio = self.smaw.Current.Value / self.sma.Current.Value
if self.smalt.Update(time, volume):
if self.smalt.Current.Value != 0 and self.smaw.Current.Value != 0:
self.volume_ratiol = self.smaw.Current.Value / self.smalt.Current.Value
if self.fast.Update(time, value) and self.slow.Update(time, value):
fast = self.fast.Current.Value
slow = self.slow.Current.Value
#self.is_uptrend = fast > slow * self.tolerance
self.is_uptrend = (fast > (slow * self.tolerance)) and (value > (fast * self.tolerance))
if self.is_uptrend:
self.scale = (fast - slow) / ((fast + slow) / 2.0)
class FScore(object):
def __init__(self, netincome, operating_cashflow, roa_current,
roa_past, issued_current, issued_past, grossm_current, grossm_past,
longterm_current, longterm_past, curratio_current, curratio_past,
assetturn_current, assetturn_past):
self.netincome = netincome
self.operating_cashflow = operating_cashflow
self.roa_current = roa_current
self.roa_past = roa_past
self.issued_current = issued_current
self.issued_past = issued_past
self.grossm_current = grossm_current
self.grossm_past = grossm_past
self.longterm_current = longterm_current
self.longterm_past = longterm_past
self.curratio_current = curratio_current
self.curratio_past = curratio_past
self.assetturn_current = assetturn_current
self.assetturn_past = assetturn_past
def ObjectiveScore(self):
fscore = 0
fscore += np.where(self.netincome > 0, 1, 0)
fscore += np.where(self.operating_cashflow > 0, 1, 0)
fscore += np.where(self.roa_current > self.roa_past, 1, 0)
fscore += np.where(self.operating_cashflow > self.roa_current, 1, 0)
fscore += np.where(self.longterm_current <= self.longterm_past, 1, 0)
fscore += np.where(self.curratio_current >= self.curratio_past, 1, 0)
fscore += np.where(self.issued_current <= self.issued_past, 1, 0)
fscore += np.where(self.grossm_current >= self.grossm_past, 1, 0)
fscore += np.where(self.assetturn_current >= self.assetturn_past, 1, 0)
return fscore