Market Bias Indicator

Just wanted to share this indicator I've been working on in case anyone sees any value. The basic idea is to determine trend while reducing the false positives in moving averages. So instead of caring about whether a price is above or below a MA, it looks at the relationship between the mean of the price for n periods vs the mean of the MA for the same period.

def get_bias(ma,pc,context):
ma = ma[-context.bias_lookback:]# array of moving average values
pc = pc[-context.bias_lookback:]# array of price close values
ma_mean = np.mean(ma)# mean of moving averages
pc_mean = np.mean(pc)# mean of price values
if ma_mean > pc_mean:# determines down bias and gets strength
strength = ma_mean - pc_mean
return 2,strength #short

if pc_mean > ma_mean:# determines up bias and gets strength
strength = pc_mean - ma_mean
return 1,strength #long


By returning the strength of the bias (up or down), we can place long and short trades for the given bias and adjust the shares according to the strength. So if the long bias is low, then we only risk a few shares and vice versa for a strong bias.

43
Total Returns
--
Alpha
--
Beta
--
Sharpe
--
Sortino
--
Max Drawdown
--
Benchmark Returns
--
Volatility
--
 Returns 1 Month 3 Month 6 Month 12 Month
 Alpha 1 Month 3 Month 6 Month 12 Month
 Beta 1 Month 3 Month 6 Month 12 Month
 Sharpe 1 Month 3 Month 6 Month 12 Month
 Sortino 1 Month 3 Month 6 Month 12 Month
 Volatility 1 Month 3 Month 6 Month 12 Month
 Max Drawdown 1 Month 3 Month 6 Month 12 Month
import talib
from datetime import datetime
import numpy as np

def initialize(context):
context.max_notional = 100000

context.stock = symbol('spy')
set_benchmark(symbol('spy'))
context.lookback = 500

context.charttype = '1d'

context.max_lvrg = 3

context.bias_lookback = 30
context.trackbias = []

context.trackbias_strength = []

context.biasSMA = 200 #Moving Average value as input to bias calc

context.avgup = []
context.avgdown = []

context.collectionperiod = 100

context.shares_per_strength = 1000

def get_bias(ma,pc,context):
ma = ma[-context.bias_lookback:]# array of moving average values
pc = pc[-context.bias_lookback:]# array of price close values
ma_mean = np.mean(ma)# mean of moving averages
pc_mean = np.mean(pc)# mean of price values
if ma_mean > pc_mean:# determines down bias and gets strength
strength = ma_mean - pc_mean
return 2,strength #short

if pc_mean > ma_mean:# determines down bias and gets strength
strength = pc_mean - ma_mean
return 1,strength #long

# Will be called on every trade event for the securities you specify.
def handle_data(context, data):

prices_close = history(context.lookback, context.charttype, 'close_price')
prices_close = list(prices_close.values.flatten())
prices = history(context.lookback, context.charttype, 'price')[context.stock]

short_SMA = talib.EMA(prices, timeperiod=context.biasSMA)
short_SMA = short_SMA.tolist()
bias = get_bias(short_SMA,prices_close,context)

try:
if bias[0] == 1:
record(bias_up = bias[1])
context.trackbias.append(bias[0])
context.trackbias_strength.append(bias[1])

context.avgup.append(bias[1])
if bias[0] == 2:
record(bias_down = bias[1])
context.trackbias.append(bias[0])
context.trackbias_strength.append(bias[1])

except:
pass

if bias != None and len(context.trackbias) >= 2:
if context.trackbias[-1] == 1:
#shares adjusted per the strenght of up bias
amnt = context.trackbias_strength[-1]*context.shares_per_strength
order_target(context.stock,amnt)

if context.trackbias[-1] == 2:
#shares adjusted per the strenght of down bias
amnt = context.trackbias_strength[-1]*context.shares_per_strength
order_target(context.stock,-amnt)

There was a runtime error.
5 responses

things to consider

36
Total Returns
--
Alpha
--
Beta
--
Sharpe
--
Sortino
--
Max Drawdown
--
Benchmark Returns
--
Volatility
--
 Returns 1 Month 3 Month 6 Month 12 Month
 Alpha 1 Month 3 Month 6 Month 12 Month
 Beta 1 Month 3 Month 6 Month 12 Month
 Sharpe 1 Month 3 Month 6 Month 12 Month
 Sortino 1 Month 3 Month 6 Month 12 Month
 Volatility 1 Month 3 Month 6 Month 12 Month
 Max Drawdown 1 Month 3 Month 6 Month 12 Month
# https://www.quantopian.com/posts/market-bias-indicator

import talib
from datetime import datetime
import numpy as np
import statsmodels.api as sm

def initialize(context):
context.stock = symbol('spy')
set_benchmark(symbol('spy'))
context.lookback            = 500
context.charttype           = '1d'
context.max_lvrg            = 3
context.bias_lookback1      = 5
context.bias_lookback2      = 30
context.biasSMA             = 400 # Moving Average value as input to bias calc
context.collectionperiod    = 100
context.shares_per_strength = 1000
context.trackbias           = []
context.trackbias_strength  = []
context.avgup               = []
context.avgdown             = []

def get_bias(ma,pc,context):
ma1 = ma[-context.bias_lookback1:]    # array of moving average values
pc1 = pc[-context.bias_lookback1:]    # array of price close values
ma2 = ma[-context.bias_lookback2:]
pc2 = pc[-context.bias_lookback2:]
ma_mean1 = np.mean(ma1)    # mean of moving averages
pc_mean1 = np.mean(pc1)    # mean of price values
ma_mean2 = np.mean(ma2)
pc_mean2 = np.mean(pc2)
ma_mean  = (ma_mean1 + ma_mean2) / 2               # avg of averages
pc_mean  = (pc_mean1 + pc_mean2) / 2
ma_mean  = ma_mean2                                # override, back to original, changed my mind
pc_mean  = pc_mean2
if ma_mean > pc_mean:    # determines down bias and gets strength
strength = ma_mean - pc_mean
return 2, strength   # short
elif pc_mean > ma_mean:  # determines down bias and gets strength
strength = pc_mean - ma_mean
return 1, strength   # long
return 3, 0    # same or null

def handle_data(context, data):
prices_close = history(context.lookback, context.charttype, 'close_price')
prices_close = list(prices_close.values.flatten())
prices       = history(context.lookback, context.charttype, 'price')[context.stock]

short_SMA = talib.SMA(prices, timeperiod=context.biasSMA)   # EMA changed to SMA
short_SMA = short_SMA.tolist()
bias      = get_bias(short_SMA,prices_close,context)

if bias[0] == 1:
record(bias_up = bias[1])
context.trackbias.append(bias[0])
context.trackbias_strength.append(bias[1])
context.avgup.append(bias[1])
elif bias[0] == 2:
record(bias_down = bias[1])
context.trackbias.append(bias[0])
context.trackbias_strength.append(bias[1])

if len(context.trackbias) < 2:
return

if bias[0] == 1:            # shares adjusted per the strength of up bias
amnt = context.trackbias_strength[-1]*context.shares_per_strength
order_target(context.stock,amnt)

elif bias[0] == 2:          # shares adjusted per the strength of down bias
spy_slp = slope_calc(prices.tolist()[-20:])
if spy_slp < 0:
#print bias[1]
amnt = context.trackbias_strength[-1] * context.shares_per_strength
print int(amnt)
order_target(context.stock, -amnt)
# Consider setting a max since nearly 50000 shares of SPY being shorted above.
#order_target(context.stock, max(-1000, -amnt))
elif spy_slp > 0:
shrs = context.portfolio.positions[symbol('SPY')].amount
if shrs < 0:
#order(context.stock, -1 * int(shrs * .5))    # buyback some

def slope_calc(in_list):
time = sm.add_constant(range(-len(in_list) + 1, 1))
return sm.OLS(in_list, time).fit().params[-1]  # slope


There was a runtime error.

Interesting, thanks garyha. I suspect this may have more of an intraday application - unfortunately that's hard to test out in Quantopian

@garyha... It stop working....

def handle_data(context, data):
----> prices_close = history(context.lookback, context.charttype, 'close_price')
prices_close = list(prices_close.values.flatten())
-----> prices = history(context.lookback, context.charttype, 'price')[context.stock]
prices = history(context.lookback, context.charttype, 'price')[context.stock]

Here is slightly modified Gary version for Q2

24
Total Returns
--
Alpha
--
Beta
--
Sharpe
--
Sortino
--
Max Drawdown
--
Benchmark Returns
--
Volatility
--
 Returns 1 Month 3 Month 6 Month 12 Month
 Alpha 1 Month 3 Month 6 Month 12 Month
 Beta 1 Month 3 Month 6 Month 12 Month
 Sharpe 1 Month 3 Month 6 Month 12 Month
 Sortino 1 Month 3 Month 6 Month 12 Month
 Volatility 1 Month 3 Month 6 Month 12 Month
 Max Drawdown 1 Month 3 Month 6 Month 12 Month
# Market Bias Indicator Gary
# https://www.quantopian.com/posts/market-bias-indicator

import talib
from datetime import datetime
import numpy as np
import statsmodels.api as sm

def initialize(context):
context.stock = symbol('spy')
set_benchmark(symbol('spy'))
context.lookback            = 500
context.charttype           = '1d'
context.max_lvrg            = 3
context.bias_lookback1      = 5
context.bias_lookback2      = 30
context.biasSMA             = 400 # Moving Average value as input to bias calc
context.collectionperiod    = 100
context.shares_per_strength = 1000
context.trackbias           = []
context.trackbias_strength  = []
context.avgup               = []
context.avgdown             = []

def get_bias(ma,pc,context):
ma1 = ma[-context.bias_lookback1:]    # array of moving average values
pc1 = pc[-context.bias_lookback1:]    # array of price close values
ma2 = ma[-context.bias_lookback2:]
pc2 = pc[-context.bias_lookback2:]
ma_mean1 = np.mean(ma1)    # mean of moving averages
pc_mean1 = np.mean(pc1)    # mean of price values
ma_mean2 = np.mean(ma2)
pc_mean2 = np.mean(pc2)
ma_mean  = (ma_mean1 + ma_mean2) / 2               # avg of averages
pc_mean  = (pc_mean1 + pc_mean2) / 2
ma_mean  = ma_mean2                                # override, back to original, changed my mind
pc_mean  = pc_mean2
if ma_mean > pc_mean:    # determines down bias and gets strength
strength = ma_mean - pc_mean
return 2, strength   # short
elif pc_mean > ma_mean:  # determines down bias and gets strength
strength = pc_mean - ma_mean
return 1, strength   # long
return 3, 0    # same or null

if get_open_orders():  return

prices_close = data.history(context.stock, 'price',context.lookback, context.charttype)
prices_close = list(prices_close.values.flatten())
prices       = data.history(context.stock, 'price',context.lookback, context.charttype)

short_SMA = talib.SMA(prices, timeperiod=context.biasSMA)   # EMA changed to SMA
short_SMA = short_SMA.tolist()
bias      = get_bias(short_SMA,prices_close,context)

if bias[0] == 1:
record(bias_up = bias[1])
context.trackbias.append(bias[0])
context.trackbias_strength.append(bias[1])
context.avgup.append(bias[1])
elif bias[0] == 2:
record(bias_down = bias[1])
context.trackbias.append(bias[0])
context.trackbias_strength.append(bias[1])

if len(context.trackbias) < 2:
return

if bias[0] == 1:            # shares adjusted per the strength of up bias
amnt = context.trackbias_strength[-1]*context.shares_per_strength
order_target(context.stock,amnt)

elif bias[0] == 2:          # shares adjusted per the strength of down bias
spy_slp = slope_calc(prices.tolist()[-20:])
if spy_slp < 0:
#print bias[1]
amnt = context.trackbias_strength[-1] * context.shares_per_strength
print int(amnt)
order_target(context.stock, -amnt)
# Consider setting a max since nearly 50000 shares of SPY being shorted above.
#order_target(context.stock, max(-1000, -amnt))
elif spy_slp > 0:
shrs = context.portfolio.positions[symbol('SPY')].amount
if shrs < 0: