python QUANTOPIAN ALGO模板（无管道）

Posted 2021-05-09

"""
This is a template algorithm on Quantopian for you to adapt and fill in.
"""
import pandas as pd
import numpy as np
 
def initialize(context):
    """
    Called once at the start of the algorithm.
    """   
    context.strategies = {
        'RS0001': {'symbls': symbols('HYS', 'MBB', 'HYMB'), 'prices': 'yahoo',
                   'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 1,
                   'frequency': 'M',
                   'cash_proxy': 'CASHX', 'risk_free': 0}}
    
    name = 'RS0001'

    context.symbols = context.strategies[name]['symbls']
    context.n_top = context.strategies[name]['n_top']
    context.frequency = context.strategies[name]['frequency']
    context.rs_lookback = context.strategies[name]['rs_lookback']
    context.risk_lookback = context.strategies[name]['risk_lookback']    
    context.cash_proxy = context.strategies[name]['cash_proxy']
    context.risk_free = context.strategies[name]['risk_free']
    
    context.tickers = context.symbols[:]
    if context.cash_proxy != 'CASHX':
        context.tickers += context.cash_proxy
    if not isinstance(context.risk_free, int):
        context.tickers += context.risk_free
    context.tickers = list(set(context.tickers))    
    
    
    if context.frequency == 'M':
        factor = 25
    else:
        factor = 1
                      
    context.max_lookback = max(context.rs_lookback, context.risk_lookback) * factor    
    
    # Rebalance at monthend.
    schedule_function(rebalance, date_rules.month_end(days_offset=0))
 
def before_trading_start(context, data):
    """
    Called every day before market open.
    """
    context.prices = data.history(context.tickers, 'price', context.max_lookback, '1d')
    
    if context.frequency == 'M':
        context.prices = context.prices.resample('M').last()
    
     
def assign_weights(context, data):
    """
    Assign weights to securities that we want to order.
    """
    # should be -2 to get previos month's return
    returns = context.prices[context.symbols].pct_change(context.rs_lookback).iloc[-1]

    if isinstance(context.risk_free, int):
        excess_returns = returns
    else:
        risk_free_returns = context.prices[context.risk_free].pct_change(context.rs_lookback).iloc[-1]
        excess_returns = returns - risk_free_returns.values

    # relative strength ranking
    ranked = excess_returns.rank(ascending=False, method='dense')
    # elligibility rule - top n_top ranked securities
    elligible = ranked <= context.n_top

    # equal weight allocations
    elligible = elligible / elligible.sum()

    # downside protection
    absolute_momentum = context.prices[context.symbols].pct_change(context.risk_lookback).iloc[-1]
    absolute_momentum_rule = absolute_momentum > 0
    weights = pd.Series(0., index=context.prices.columns)
    weights[context.symbols] = elligible * absolute_momentum_rule
    if context.cash_proxy != 'CASHX':
        weights[context.cash_proxy] = 1 - weights[context.symbols].sum()
                      
    return weights
 
def rebalance(context,data):
    """
    Execute orders according to our schedule_function() timing. 
    """
    if np.isnan(context.prices.iloc[0]).any():
        return
    
    weights = assign_weights(context, data)
    
    log.info ('WEIGHTS : {} SUM = {}'.format([(weights.index[i].symbol, v.round(2)) for i,v in enumerate(weights)],
                                            weights.sum())) 
    
    for i,v in enumerate(weights):
        order_target_percent(weights.index[i], v)