Back to Community
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.

Clone Algorithm
43
Loading...
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

Clone Algorithm
36
Loading...
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_target(context.stock, 0)    # buyback
                #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]

Vlademir,

Here is slightly modified Gary version for Q2

Clone Algorithm
24
Loading...
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             = []
    schedule_function(trade, date_rules.every_day(), time_rules.market_close(minutes = 15))

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 trade(context, data):
    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:
                order_target(context.stock, 0)    # buyback
                #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.

thanks Vladimir.... having alot of difficulty in using this new api... and codes... which are not working anymore... its not backward compatible.... with codes that have work in the the pass....