Back to Community
Genetic Algorithm to predict best SMA to trade off of

Hi I'm new to Python and Quantopian so you can understand my frustration. However, I'm not new to coding.

Here's my pseudo code:

create method to create individual

each individual should only have two randomly SMA days to trade off of

create method to initialize the population. probably about six individuals

create the fitness function to test each individual.They will be evaluated by the percentage gains they make. Only take the top 4 individuals.

Create two more individuals and add them to the list. Breed them with a chance to mutate.

Rinse and Repeat until I have the best two SMA's to trade off of.

One major question I have, how would I store the past prices to run each individual through the fitness test?

Thanks in advance.

17 responses

Two options:
1) call up the history function, documentation of how to use it is in the API doc.
2)create an empty array : APPLprices = []
then use APPLprices.append() in handle_data to append each new bar of data to the array

The second method would not be best practice as it could grow huge. But is more efficient if you are referencing the prices many times, in this case I would limit the array length so it doesn't get too large.

Here's the code I ended up with. It is very similar to the evolutionary strategy, except that I am using only one stock to find it's best SMA days to trade with. I can't seem to print anything to the log file. Any help would be much appreciated,

# 1. Generate 10 random sma strategies  
# 2. Run these strategies for 365 days  
# 3. Pick the best of the strategies that have been run, and generate 5 new strategies derived from the top performers  
# 4. Continue with this process, and make trades based off the strategy that performed the best in the past

import random  
from collections import deque

# Put any initialization logic here.  The context object will be passed to  
# the other methods in your algorithm.  
def initialize(context):  
    context.security = symbol('PTBI')  #security being bought  
    set_commission(commission.PerTrade(cost=0))  #commission prices  
    context.target_duration = 365  # number of days to test individual  
    context.num_parents = 1 # number of parents each new individual HAS  
    context.generation_size = 10  
    context.frames = deque(maxlen=500)  #stores prices here  
    context.individuals= []  # all individuals are stored here  
    new_generation(context)  #call new generation  


#creates a new generation based off of size context.generation_size  
def new_generation(context):  
    possible_parents = []  
    for indi in context.individuals:  
        if indi['run_duration'] >= context.target_duration:  
            possible_parents.append(indi)  
    parents = sorted(possible_parents, key = lambda k: k['value'])[-context.num_parents:]  
    current_size = context.generation_size  
    if len(possible_parents) < context.generation_size:  
       current_size = len(possible_parents)  
    for i in range(current_size):  
        # create new individual with one gene from parent and a new random gene  
        sma1 = random.randint(1,200)  
        sma2 = [parents[len(parents - i)]]['sma2']  
        #store each individuals genes, run_duration, its value, and position  
        context.individuals.append({'sma1':sma1, 'sma2':sma2, 'run_duration':0, 'value':100,'current_longs':[]})  
    return None  


#run the two crossovers against the stock price  
def run_strategy(key, context, data):  
    strat = context.individuals[key]  
    for sid in data:  
        #if SMA'S are equal and price is above and no shares have been bought  
        if data[sid].mavg(strat['sma1']) == data[sid].mavg(strat['sma2']) and                      strat['current_longs'] == 0:  
           if data[sid].mavg(strat['sma1']) < data[sid].price:  
              strat['current_longs'].append(sid)  
        elif data[sid].mavg(strat['sma1']) == data[sid].mavg(strat['sma2']) and                      strat['current_longs'] == 1:   strat['current_longs'].remove(sid)  
    return None

def update_returns(key,context,data):  
    strat = context.individuals[key]  
    daily_total_returns = 0  
    for sid in data:  
        if sid in strat['current_longs']:  
            daily_total_returns += data[sid].returns()  
    return None


# Will be called on every trade event for the securities you specify.  
def handle_data(context, data):  
    context.frames.append(data) #update bars  
    top_returner =[]  
    if len(context.frames) == 70:  
        new_gen = 1  
        #find top performer  
        tested_strats = []  
        for strat in context.individuals:  
            if strat['run_duration'] >= context.target_duration:  
                tested_strats.append(strat)  
        top_returners = sorted(tested_strats, key =lambda k: k['value'])  

        #check each individual to see if it needs to be run again  
        for i in range(len(context.individuals)):  
            if context.individuals[i]['run_duration'] < context.target_duration or context.individuals[i] == top_returner:  
                #update returns  
                if context.individuals[i] != top_returner:  
                    update_returns(i, context, data)  
                    context.individuals[i]['run_duration'] += 1  
                    new_gen = 0  
                run_strategy(i, context, data)  
        if new_gen == 1:  
            log.info('sma1: '+str(top_returner['sma1']))  
            log.info('sma2: '+str(top_returner['sma2']))

            new_generation(context) # Create new generation  
        longs = []  
        context.portfolio.portfolio_value  
        # Make trades based on the top performer  
        # Buy and sell shares in 2:1 leverage with amount proportional to our starting cash  
        if top_returner:  
            for sid in data:  
                if sid in top_returner['current_longs']:  
                    longs.append(sid)

            for sid in data:  
                if sid in longs:  
                    order_target_percent(sid, 1.0)  
                else:  
                    order_target(sid, 0)  
    try: top_returner['value']  
    except: top_returner['value'] = 0  
    record(num_individuals=len(context.individuals), top_value=top_returner['value'])  

If the logs aren't getting printed that means that section of code isn't getting triggered (new_gen==1). Try using the built-in debugger in the IDE to step through the lines and analyze the values: https://www.quantopian.com/help#debugger

I did a quick test and it looks like this top condition isn't getting evaluated to true, which is why further down the logs are not printing.

if len(context.frames) == 70  
Disclaimer

The material on this website is provided for informational purposes only and does not constitute an offer to sell, a solicitation to buy, or a recommendation or endorsement for any security or strategy, nor does it constitute an offer to provide investment advisory services by Quantopian. In addition, the material offers no opinion with respect to the suitability of any security or specific investment. No information contained herein should be regarded as a suggestion to engage in or refrain from any investment-related course of action as none of Quantopian nor any of its affiliates is undertaking to provide investment advice, act as an adviser to any plan or entity subject to the Employee Retirement Income Security Act of 1974, as amended, individual retirement account or individual retirement annuity, or give advice in a fiduciary capacity with respect to the materials presented herein. If you are an individual retirement or other investor, contact your financial advisor or other fiduciary unrelated to Quantopian about whether any given investment idea, strategy, product or service described herein may be appropriate for your circumstances. All investments involve risk, including loss of principal. Quantopian makes no guarantees as to the accuracy or completeness of the views expressed in the website. The views are subject to change, and may have become unreliable for various reasons, including changes in market conditions or economic circumstances.

I keep on getting this error:

USER ALGORITHM:103, in handle_data
log.info('sma1: '+str(top_returner['sma1']))

Which confuses me because I'm logging it so it should be a string anyways, right?

Can you share the current version of your code? I wasn't able to trigger that error in the algo above

It's still the same code. I set a break point from where you stated. It won't log the info I want it to.

@Dan,

Since the individuals are random, past performance doesn't guarantee future performance. Or am I missing something?

Bharath,
the individuals each have two genes. The first gene is random, the second is a gene taken from one of the parents.

I've been working on this algorithm for awhile now. I've been getting past most of the errors but this is one I have been stuck on for awhile. Here is the run-time error code I'm getting.

ValueError: The truth value of a Series is ambiguous. Use a.empty, a.bool(), a.item(), a.any() or a.all().
There was a runtime error on line 120.

Below is all the code:

import random  
from collections import deque

# Put any initialization logic here.  The context object will be passed to  
# the other methods in your algorithm.  
def initialize(context):  
    context.security = symbol('AAPL')  #security being bought  
    set_commission(commission.PerTrade(cost=0))  #commission prices  
    context.target_duration = 365  # number of days to test individual  
    context.num_parents = 2 # number of parents each new individual HAS  
    context.generation_size = 3  
    context.counter = 0  #stores prices here  
    context.individuals = [] # all individuals are stored here  
    parents = [] # parents stored here  
    new_generation(context)  # call new generation  
#creates a new generation based off of size context.generation_size

def new_generation(context):  
    possible_parents = []  
    for indi in context.individuals:  
        log.info(context.individuals[indi])  
        if indi['run_duration'] >= context.target_duration:  
            possible_parents.append(indi)  
    parents = sorted(possible_parents, key = lambda k: k['value'])[-context.num_parents:]  
    for i in range(context.generation_size):  
        # create new individual with one gene from parent and a new random gene  
        sma1 = random.randint(1,200)  
        if len(parents) == 0:  
            sma2 = random.randint(1,200)  
        else:  
            sma2 = parents[random.randrange(len(context.num_parents))]['sma2']  
            # parents[len(parents - i)]['sma2']  
        #store each individuals genes, run_duration, its value, and position  
        context.individuals.append({'sma1':sma1, 'sma2':sma2, 'run_duration':0,             'value':100,'current_longs':[]})  
    return None  


#run the two crossovers against the stock price  
def run_strategy(key, context, data):  
    ####################replace line 67 with "sma1 > sma2"  
    strat = context.individuals[key]  
    prices = history(300, '1d', 'price')  
    sma1 = prices.tail(strat['sma1']).mean()  
    sma2 = prices.tail(strat['sma2']).mean()

    if (sma1 < sma2) and strat['current_longs'] == 0:  
        strat['current_longs'].append(sid)  
    elif (sma2 <  sma1) and strat['current_longs'] == 1:  
        strat['current_longs'].remove(sid)  
    #for sid in data:  
        #if SMA'S are equal and price is above and no shares have been bought  
    #if data[sid].mavg(strat['sma1']) == data[sid].mavg(strat['sma2']) and                      strat['current_longs'] == 0:  
       # if data[sid].mavg(strat['sma1']) < data[sid].price:  
            #  strat['current_longs'].append(sid)  
       # elif data[sid].mavg(strat['sma1']) == data[sid].mavg(strat['sma2']) and                      strat['current_longs'] == 1:   strat['current_longs'].remove(sid)  
    return None

def update_returns(key,context,data):  
    strat = context.individuals[key]  
    daily_total_returns = 0  
    for sid in data:  
        if sid in strat['current_longs']:  
            daily_total_returns += data[sid].returns()  
    return None


# Will be called on every trade event for the securities you specify.  
def handle_data(context, data):  
    #prices = history(300, '1d', 'price')  
    #sma1 = prices.tail(context.individuals['sma1']).mean()  
    #sma2 = prices.tail(smas['sma2']).mean()  
    context.counter += 1  
    top_returner =[] #fix error  
    if context.counter >= 70:  
        new_gen = 1  
        #find top performer  
        tested_strats = []  
        for strat in context.individuals:  
            if strat['run_duration'] >= context.target_duration:  
               tested_strats.append(strat)  
        top_returners = sorted(tested_strats, key =lambda k: k['value'])  
        if top_returners:  
           top_returner = top_returners[-1]  

        #check each individual to see if it needs to be run again  
        for i in range(len(context.individuals)):  
            if context.individuals[i]['run_duration'] < context.target_duration or                     context.individuals[i] == top_returner:  
                #update returns  
                if context.individuals[i] != top_returner:  
                    update_returns(i, context, data)  
                    context.individuals[i]['run_duration'] += 1  
                    new_gen = 0  
                run_strategy(i, context, data)  
        if new_gen == 1:  
            log.info('sma1: '+ str(top_returner['sma1']))  
            log.info('sma2: '+ str(top_returner['sma2']))  
            new_generation(context) # Create new generation  
        longs = []  
        context.portfolio.portfolio_value  
       # Make trades based on the top performer  
        # Buy and sell shares in 2:1 leverage with amount proportional to our starting cash  
        if top_returner:  
            for sid in data:  
               if sid in top_returner['current_longs']:  
                  longs.append(sid)

            for sid in data:  
                if sid in longs:  
                    order_target_percent(sid, 1.0)  
                else:  
                    order_target(sid, 0)  
        try: top_returner['value']  
        except: top_returner['value'] = 0  
        record(num_individuals=len(context.individuals), top_value=top_returner['value'])  

prices.tail(strat['sma1']).mean() returns a series of means and after that your do a comparison sma1 < sma2 which compares an array with array. Do you actually want to do a compare of every sma1 and sma2 per stock? That has to be solved in a for loop like sometimg like this

x = [stock for stock in val if val[stock].sma1 < val[stock].sma2] or with a normal for loop to walk over all the elements

I was debugging your code because I found it interesting, and after I fixed the first 3 I realized that the algorithm didn't work, and no trades were executed. Looks like there are some logical bugs too. I suggest you rewrite the whole thing from scratch, and test as you go along. If you want I can give you the code I fixed, but I gave up after debugging and seeing that your top_returners array remained totally empty throughout.

import random  
from collections import deque

# Put any initialization logic here.  The context object will be passed to  
# the other methods in your algorithm.  
def initialize(context):  
    context.security = symbol('AAPL')  #security being bought  
    set_commission(commission.PerTrade(cost=0))  #commission prices  
    context.target_duration = 365  # number of days to test individual  
    context.num_parents = 2 # number of parents each new individual HAS  
    context.generation_size = 3  
    context.counter = 0  #stores prices here  
    context.individuals = [] # all individuals are stored here  
    parents = [] # parents stored here  
    new_generation(context)  # call new generation  
#creates a new generation based off of size context.generation_size

def new_generation(context):  
    possible_parents = []  
    for indi in context.individuals:  
        log.info(context.individuals[indi])  
        if indi['run_duration'] >= context.target_duration:  
            possible_parents.append(indi)  
    parents = sorted(possible_parents, key = lambda k: k['value'])[-context.num_parents:]  
    for i in range(context.generation_size):  
        # create new individual with one gene from parent and a new random gene  
        sma1 = random.randint(1,200)  
        if len(parents) == 0:  
            sma2 = random.randint(1,200)  
        else:  
            sma2 = parents[random.randrange(len(context.num_parents))]['sma2']  
            # parents[len(parents - i)]['sma2']  
        #store each individuals genes, run_duration, its value, and position  
        context.individuals.append({'sma1':sma1, 'sma2':sma2, 'run_duration':0, 'value':100,'current_longs':[]})  
    return None  


#run the two crossovers against the stock price  
def run_strategy(key, context, data):  
    ####################replace line 67 with "sma1 > sma2"  
    strat = context.individuals[key]  
    # prices = history(300, '1d', 'price')  
    # sma1 = prices.tail(strat['sma1']).mean()  
    # sma2 = prices.tail(strat['sma2']).mean()  
    sma1prices = history(strat['sma1'], '1d', 'price')  
    sma2prices = history(strat['sma2'], '1d', 'price')  
    for sid in data:  
        sma1 = sma1prices[sid].mean()  
        sma2 = sma2prices[sid].mean()  
        if (sma1 < sma2) and not(sid in strat['current_longs']):  
            strat['current_longs'].append(sid)  
        elif (sma2 <  sma1) and (sid in strat['current_longs']):  
            strat['current_longs'].remove(sid)  
    #for sid in data:  
        #if SMA'S are equal and price is above and no shares have been bought  
    #if data[sid].mavg(strat['sma1']) == data[sid].mavg(strat['sma2']) and                      strat['current_longs'] == 0:  
       # if data[sid].mavg(strat['sma1']) < data[sid].price:  
            #  strat['current_longs'].append(sid)  
       # elif data[sid].mavg(strat['sma1']) == data[sid].mavg(strat['sma2']) and                      strat['current_longs'] == 1:   strat['current_longs'].remove(sid)  
    return None

def update_returns(key,context,data):  
    strat = context.individuals[key]  
    daily_total_returns = 0  
    for sid in data:  
        if sid in strat['current_longs']:  
            daily_total_returns += data[sid].returns()  
    return None


# Will be called on every trade event for the securities you specify.  
def handle_data(context, data):  
    #prices = history(300, '1d', 'price')  
    #sma1 = prices.tail(context.individuals['sma1']).mean()  
    #sma2 = prices.tail(smas['sma2']).mean()  
    context.counter += 1  
    top_returner =[] #fix error  
    if context.counter >= 70:  
        new_gen = 1  
        #find top performer  
        tested_strats = []  
        for strat in context.individuals:  
            if strat['run_duration'] >= context.target_duration:  
               tested_strats.append(strat)  
        top_returners = sorted(tested_strats, key =lambda k: k['value'])  
        if top_returners:  
           top_returner = top_returners[-1]  

        #check each individual to see if it needs to be run again  
        for i in range(len(context.individuals)):  
            if context.individuals[i]['run_duration'] < context.target_duration or context.individuals[i] == top_returner:  
                #update returns  
                if context.individuals[i] != top_returner:  
                    update_returns(i, context, data)  
                    context.individuals[i]['run_duration'] += 1  
                    new_gen = 0  
                run_strategy(i, context, data)  
        if new_gen == 1:  
            log.info('sma1: '+ str(top_returner['sma1']))  
            log.info('sma2: '+ str(top_returner['sma2']))  
            new_generation(context) # Create new generation  
        longs = []  
        context.portfolio.portfolio_value  
       # Make trades based on the top performer  
        # Buy and sell shares in 2:1 leverage with amount proportional to our starting cash  
        if top_returner:  
            for sid in data:  
               if sid in top_returner['current_longs']:  
                  longs.append(sid)

            for sid in data:  
                if sid in longs:  
                    order_target_percent(sid, 1.0)  
                else:  
                    order_target(sid, 0)  
        try: top_returner['value']  
        except: top_returner['value'] = 0  
        record(num_individuals=len(context.individuals), top_value=top_returner['value'])  

Here's a version that works with history. This was just modified from the original strategy I made, and it doesn't account for any changes that were introduced in this post, but I'm guessing it can be easily modified to work. This strategy isn't really that good and isn't intended for real use. I think the best way to go about this is to run a backtest using this algorithm, and then have the backtest log what the top strategy was at the end. That could then be used as the signal. I ran a long backtest just because I was curious. I tweaked some parameters like the leverage (should have been 1 and not 2). Also note that this thing makes so many trades that commission kills it, so I just turned commission off (also because the fitness function doesn't take into account # of trades, or commission). Also, here is my original post for anyone who was curious.

Clone Algorithm
304
Loading...
Backtest from to with initial capital
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
# Backtest ID: 55b654898740580c7689a23b
There was a runtime error.
Disclaimer

The material on this website is provided for informational purposes only and does not constitute an offer to sell, a solicitation to buy, or a recommendation or endorsement for any security or strategy, nor does it constitute an offer to provide investment advisory services by Quantopian. In addition, the material offers no opinion with respect to the suitability of any security or specific investment. No information contained herein should be regarded as a suggestion to engage in or refrain from any investment-related course of action as none of Quantopian nor any of its affiliates is undertaking to provide investment advice, act as an adviser to any plan or entity subject to the Employee Retirement Income Security Act of 1974, as amended, individual retirement account or individual retirement annuity, or give advice in a fiduciary capacity with respect to the materials presented herein. If you are an individual retirement or other investor, contact your financial advisor or other fiduciary unrelated to Quantopian about whether any given investment idea, strategy, product or service described herein may be appropriate for your circumstances. All investments involve risk, including loss of principal. Quantopian makes no guarantees as to the accuracy or completeness of the views expressed in the website. The views are subject to change, and may have become unreliable for various reasons, including changes in market conditions or economic circumstances.

Thank you Gus for posting this. It has helped immensely. One thing though is that I keep on getting a key error and I can't really see what is causing it. I'
m pumping out lots of individuals but nothing is being bought. I don't quite understand what is going on here. Any help is appreciated!

Clone Algorithm
11
Loading...
Backtest from to with initial capital
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
# Backtest ID: 55bf03abd350750c691457c7
There was a runtime error.

What are you trying to do differently from before? Also, where is the KeyError popping up?

On this algorithm I'm not shorting any stocks and each species only has two traits. On lines 89 and 90, once when I try to log the top returners I get a KeyError. I can't really pinpoint what is causing the error.

OK, it looks like you have the line if top_returners:. The algorithm stops here, and this prevents it from ever actually testing any strategy. You can test this by printing anything, for example print 'hello', within that if statement. You will see "hello" never gets logged in the console, so that line never executes. You will want to fix this by allowing the strategies to run on the first day when there is no top returner, or seeing how I do it in my code. Hope that helps.

Okay I think I have everything figured out. I can run the algorithm just fine but I'm wondering if there are any logical errors within it. My "top values" never change. Once again, thanks for everyone's help on this algorithm.

Clone Algorithm
11
Loading...
Backtest from to with initial capital
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
# Backtest ID: 55c10989058ffd0c5f141d15
There was a runtime error.