python BACKTEST RS_DM或PMA STEP_BY_STEP

Posted 2021-05-08

import matplotlib.pyplot as plt
import pandas as pd
from datetime import datetime, timezone
import pytz

import ffn

from fintools import get_DataArray,compute_weights_RS_DM,compute_weights_PMA,\
                     Parameters, show_return_table, show_annual_returns, \
                     endpoints, backtest

start = datetime(2000, 1, 1, 0, 0, 0, 0, pytz.utc)
end = datetime.today().replace(tzinfo=timezone.utc)

# CHOOSE ONLY ONE

strategies = {
    #     'RS0001': { 'assets': ['VCVSX','VWEHX','VFIIX','FGOVX','VWAHX'],
    #                'start':start, 'end':end,
    #                'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 2, 'frequency': 'M',
    #               'cash_proxy': 'CASHX', 'risk_free': 0},
    #     'RS0002': {'assets': ['MMHYX','FAGIX','VFIIX'],
    #                'start':start, 'end':end,
    #                'rs_lookback': 3, 'risk_lookback': 2, 'n_top': 1, 'frequency': 'M',
    #               'cash_proxy': 'CASHX', 'risk_free': 0},
    #     'RS0003': {'assets': ['MMHYX','FAGIX','VFIIX'],
    #                'start':start, 'end':end,
    #                'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 1, 'frequency': 'Q',
    #               'cash_proxy': 'CASHX', 'risk_free': 0},

    'DM0001': {'assets': ['VCVSX', 'VWINX', 'VWEHX', 'VGHCX', 'VFIIX', 'VWAHX', 'FGOVX', 'FFXSX'],
               'start': start, 'end': end,
               'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 3, 'frequency': 'M',
               'cash_proxy': 'VUSTX', 'risk_free': 0},

    #     'DM0002': {'assets': ['VCVSX','VUSTX','VWEHX','VFIIX','VGHCX','FRESX'],
    #                'start':start, 'end':end,
    #                'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 5, 'frequency': 'M',
    #               'cash_proxy': 'VFIIX', 'risk_free': 'FFXSX'},
}

# *****************************************************************
# Load historical data
# ******************************************************************

name = [i for i in strategies.items()][0][0]
p = Parameters(strategies[name])

cash_proxy = p.cash_proxy
risk_free = p.risk_free

rs_lookback = p.rs_lookback
risk_lookback = p.risk_lookback
n_top = p.n_top
# allocations = p.allocations

assets = p.assets

# get data
tickers = assets.copy()
if cash_proxy != 'CASHX':
    tickers = list(set(tickers + [cash_proxy]))
if isinstance(risk_free, str):
    tickers = list(set(tickers + [risk_free]))

da = get_DataArray(tickers, start, end)
# don't forget to ffill()
data = da.to_pandas().transpose(1, 2, 0).ffill()

data1 = data.copy()[:, :, 'adj close']
inception_dates = pd.DataFrame([data1[ticker].first_valid_index() for ticker in data1.columns],
                               index=data1.keys(), columns=['inception'])

# print (inception_dates)

prices = data.copy()[:, :, 'adj close'].dropna()

end_points = endpoints(period=p.frequency, trading_days=prices.index)
prices_m = prices.loc[end_points]

# print(prices_m[:3])

returns = prices_m[p.assets].pct_change(p.rs_lookback)[p.rs_lookback:]

if isinstance(p.risk_free, int):
    excess_returns = returns
else:
    risk_free_returns = prices_m[p.risk_free].pct_change(p.rs_lookback)[p.rs_lookback:]
    excess_returns = returns.subtract(risk_free_returns, axis=0).dropna()

absolute_momentum = prices_m[p.assets].pct_change(p.risk_lookback)[p.risk_lookback:]
absolute_momentum_rule = absolute_momentum > 0
rebalance_dates = excess_returns.index.join(absolute_momentum_rule.index, how='inner')

# relative strength ranking
ranked = excess_returns.loc[rebalance_dates][p.assets].rank(ascending=False, axis=1, method='dense')
# elligibility rule - top n_top ranked securities
elligible = ranked[ranked <= p.n_top] > 0

# equal weight allocations
elligible = elligible.multiply(1. / elligible.sum(1), axis=0)

# downside protection
weights = pd.DataFrame(0., index=elligible.index, columns=prices.columns)
if p.cash_proxy == 'CASHX':
    weights[p.cash_proxy] = 0
    prices[p.cash_proxy] = 1.
weights[p.assets] = (elligible * absolute_momentum_rule).dropna()
weights[p.cash_proxy] += 1 - weights[p.assets].sum(axis=1)

# backtest

p_value, p_holdings, p_weights = backtest(prices, weights, 10000., offset=0, commission=10.)

# p_value.plot(figsize=(15, 10), grid=True, legend=True, label=name)

ffn.calc_perf_stats(p_value).display()

# show_return_table(p_value)
#
# show_annual_returns(p_value)

import pandas_datareader.data as pdr
import matplotlib.pyplot as plt
import pandas as pd
from datetime import datetime, timezone
import pytz
import talib
import itable
import ffn

from fintools import get_DataArray,compute_weights_RS_DM,compute_weights_PMA,\
                     Parameters, show_return_table, show_annual_returns, \
                     endpoints, backtest

start = datetime(2000, 1, 1, 0, 0, 0, 0, pytz.utc)
end = datetime.today().replace(tzinfo=timezone.utc)

# CHOOSE ONLY ONE

strategies = {
    'PMA001': {'assets': ['VCVSX', 'VFIIX'],
               'start':start, 'end':end,
               'risk_lookback': 3, 'frequency': 'M', 'allocations': [0.6, 0.4],
              'cash_proxy': 'VUSTX'},
    # 'PMA002': {'assets': ['VCVSX', 'VWINX', 'VWEHX'],
    #            'start':start, 'end':end,
    #            'risk_lookback': 3, 'frequency': 'M', 'allocations': [0.6, 0.2, 0.2],
    #           'cash_proxy': 'VUSTX'},
    # 'PMA003': {'assets': ['VCVSX', 'FAGIX', 'VGHCX'],
    #            'start':start, 'end':end,
    #            'risk_lookback': 2, 'frequency': 'M', 'allocations': [1./3., 1./3., 1./3.],
    #           'cash_proxy': 'VUSTX'}
    }

# *****************************************************************
# Load historical data
# ******************************************************************

name = [i for i in strategies.items()][0][0]
p = Parameters(strategies[name])

cash_proxy = p.cash_proxy

rs_lookback = None
risk_lookback = p.risk_lookback
risk_free = None
allocations = p.allocations

assets = p.assets

# get data
tickers = assets.copy()
if cash_proxy != 'CASHX':
    tickers = list(set(tickers + [cash_proxy]))
if isinstance(risk_free, str):
    tickers = list(set(tickers + [risk_free]))

da = get_DataArray(tickers, start, end)
# don't forget to ffill()
data = da.to_pandas().transpose(1, 2, 0).ffill()

data1 = data.copy()[:, :, 'adj close']
inception_dates = pd.DataFrame([data1[ticker].first_valid_index() for ticker in data1.columns],
                               index=data1.keys(), columns=['inception'])

# print (inception_dates)

prices = data.copy()[:, :, 'adj close'].dropna()

end_points = endpoints(period=p.frequency, trading_days=prices.index)
prices_m = prices.loc[end_points]

# print(prices_m[:3])

# elligibility rule
SMA = prices_m.rolling(p.risk_lookback).mean().dropna()
rebalance_dates = SMA.index
rule = prices_m.loc[rebalance_dates][p.assets] > SMA[p.assets]

# fixed weight allocation
weights = p.allocations * rule

# downside protection
weights[p.cash_proxy] = 1 - weights[p.assets].sum(axis=1)

# backtest
p_value, p_holdings, p_weights = backtest(prices, weights, 10000., offset=0, commission=10.)

p_value.plot(figsize=(15, 10), grid=True, legend=True, label=name)

# p_value.plot(figsize=(15, 10), grid=True, legend=True, label=name)

ffn.calc_perf_stats(p_value).display()

# show_return_table(p_value)
#
# show_annual_returns(p_value)