交易策略回测中存在循环问题

发布于 2025-01-09 20:09:14 字数 5705 浏览 1 评论 0原文

我目前正在尝试编写一些代码,对一个简单的交易策略进行回溯测试,包括通过时间序列价格数据进行排序,逐步拟合 ARIMA 模型,进行未来价格预测,然后在预测价格上涨时添加份额,或者如果预计价格会下跌,则出售所有累积的股票。目前,它正在返回交易预计回报的纳米值,并且似乎只是以某种方式出售。

我在下面附上了我的代码。只有一些简单的函数用于计算夏普比率,然后是用于运行回测的主要函数。

import yfinance as yf
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
from statsmodels.graphics.tsaplots import plot_acf, plot_pacf
from statsmodels.tsa.arima.model import ARIMA
import numpy as np
import seaborn as sns
from tqdm import tqdm
import pandas as pd
from statsmodels.tools.sm_exceptions import ValueWarning, HessianInversionWarning, ConvergenceWarning
import warnings

#in practice do not supress these warnings, they carry important information about the status of your model
warnings.filterwarnings('ignore', category=ValueWarning)
warnings.filterwarnings('ignore', category=HessianInversionWarning)
warnings.filterwarnings('ignore', category=ConvergenceWarning)

tickerSymbol = 'SPY'
data = yf.Ticker(tickerSymbol)

prices = data.history(start='2017-01-01', end='2019-01-01').Close
returns = prices.pct_change().dropna()

def std_dev(data):
    # Get number of observations
    n = len(data)
    # Calculate mean
    mean = sum(data) / n
    # Calculate deviations from the mean
    deviations = sum([(x - mean)**2 for x in data])
    # Calculate Variance & Standard Deviation
    variance = deviations / (n - 1)
    s = variance**(1/2)
    return s

# Sharpe Ratio From Scratch
def sharpe_ratio(data, risk_free_rate=0):
    # Calculate Average Daily Return
    mean_daily_return = sum(data) / len(data)
    print(f"mean daily return = {mean_daily_return}")
    # Calculate Standard Deviation
    s = std_dev(data)
    # Calculate Daily Sharpe Ratio
    daily_sharpe_ratio = (mean_daily_return - risk_free_rate) / s
    # Annualize Daily Sharpe Ratio
    sharpe_ratio = 252**(1/2) * daily_sharpe_ratio
    return sharpe_ratio

def run_simulation(returns, prices, amt, order, thresh, verbose=True, plot=True):
    if type(order) == float:
        thresh = None
        

    sum_list = []
    events_list = []
    sharpe_list = []
    init_amt = amt

    #go through dates
    for date, r in tqdm (returns.iloc[14:].items(), total=len(returns.iloc[14:])):
        #if you're currently holding the stock, sell it
        

        #get data til just before current date
        curr_data = returns[:date]
        
        if type(order) == tuple:
            try:
                #fit model
                model = ARIMA(curr_data, order=order).fit(maxiter=200)

                #get forecast
                pred = model.forecast()[0][0]

            except:
                pred = thresh - 1


        #if you predict a high enough return and not holding, buy stock
        if ((type(order) == float and np.random.random() < order) 
         or (type(order) == tuple and pred > thresh)):
        
            
           
            buy_price = prices.loc[date]
            events_list.append(('b', date))
            int_buy_price = int(buy_price)
            sum_list.append(int_buy_price)
            if verbose:
                print('Bought at $%s'%buy_price)
                print('Predicted Return: %s'%round(pred,4))
                print('Actual Return: %s'%(round(ret, 4)))
                print('=======================================')
            continue

        #if you predict below the threshold return, sell the stock
        if ((type(order) == float and np.random.random() < order) 
         or (type(order) == tuple and thresh > pred)
         or (order == 'last' and curr_data[-1] > 0)):
            
            sell_price = prices.loc[date]
            
            total_return = len(sum_list) * sell_price 

            ret = (total_return-sum(sum_list))/sum(sum_list)
            amt *= (1+ret)
            events_list.append(('s', date, ret))
            sharpe_list.append(ret)
            sum_list.clear()

            if verbose:
                print('Sold at $%s'%sell_price)
                print('Predicted Return: %s'%round(pred,4))
                print('Actual Return: %s'%(round(ret, 4)))
                print('=======================================')
            

                
    if verbose:
        sharpe = sharpe_ratio(sharpe_list, risk_free_rate=0)
        print('Total Amount: $%s'%round(amt,2))
        print(f"Sharpe Ratio: {sharpe}")
        
    #graph
    if plot:
    
        plt.figure(figsize=(10,4))
        plt.plot(prices[14:])

        y_lims = (int(prices.min()*.95), int(prices.max()*1.05))
        shaded_y_lims = int(prices.min()*.5), int(prices.max()*1.5)

        for idx, event in enumerate(events_list):
            plt.axvline(event[1], color='k', linestyle='--', alpha=0.4)
            if event[0] == 's':
                color = 'green' if event[2] > 0 else 'red'
                plt.fill_betweenx(range(*shaded_y_lims), 
                                  event[1], events_list[idx-1][1], color=color, alpha=0.1)

        tot_return = round(100*(amt / init_amt - 1), 2)
        sharpe = sharpe_ratio(sharpe_list, risk_free_rate=0)
        tot_return = str(tot_return) + '%'
        plt.title("%s Price Data\nThresh=%s\nTotal Amt: $%s\nTotal Return: %s"%(tickerSymbol, thresh, round(amt,2), tot_return), fontsize=20)
        plt.ylim(*y_lims)
        plt.show()
        print(sharpe)
    
    return amt

# A model with a dth difference to fit and ARMA(p,q) model is called an ARIMA process 
# of order (p,d,q). You can select p,d, and q with a wide range of methods, 
# including AIC, BIC, and empirical autocorrelations (Petris, 2009).


for thresh in [0.001]:
    run_simulation(returns, prices, 100000, (7,1,7), thresh, verbose=True)

I'm currently trying to put together some code that backtests a simple trading strategy involving sequencing through time series price data, incrementally fitting an ARIMA model, making future price predictions, and then either adding a share if the price is predicted to increase, or selling all accumulated shares if the price is predicted to go down. Currently, it's returning nan values for the projected returns from trades and appears to only be selling somehow.

I've attached my code below. There's just a few simple functions for calculating a sharpe ratio and then the main function for running backtests.

import yfinance as yf
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
from statsmodels.graphics.tsaplots import plot_acf, plot_pacf
from statsmodels.tsa.arima.model import ARIMA
import numpy as np
import seaborn as sns
from tqdm import tqdm
import pandas as pd
from statsmodels.tools.sm_exceptions import ValueWarning, HessianInversionWarning, ConvergenceWarning
import warnings

#in practice do not supress these warnings, they carry important information about the status of your model
warnings.filterwarnings('ignore', category=ValueWarning)
warnings.filterwarnings('ignore', category=HessianInversionWarning)
warnings.filterwarnings('ignore', category=ConvergenceWarning)

tickerSymbol = 'SPY'
data = yf.Ticker(tickerSymbol)

prices = data.history(start='2017-01-01', end='2019-01-01').Close
returns = prices.pct_change().dropna()

def std_dev(data):
    # Get number of observations
    n = len(data)
    # Calculate mean
    mean = sum(data) / n
    # Calculate deviations from the mean
    deviations = sum([(x - mean)**2 for x in data])
    # Calculate Variance & Standard Deviation
    variance = deviations / (n - 1)
    s = variance**(1/2)
    return s

# Sharpe Ratio From Scratch
def sharpe_ratio(data, risk_free_rate=0):
    # Calculate Average Daily Return
    mean_daily_return = sum(data) / len(data)
    print(f"mean daily return = {mean_daily_return}")
    # Calculate Standard Deviation
    s = std_dev(data)
    # Calculate Daily Sharpe Ratio
    daily_sharpe_ratio = (mean_daily_return - risk_free_rate) / s
    # Annualize Daily Sharpe Ratio
    sharpe_ratio = 252**(1/2) * daily_sharpe_ratio
    return sharpe_ratio

def run_simulation(returns, prices, amt, order, thresh, verbose=True, plot=True):
    if type(order) == float:
        thresh = None
        

    sum_list = []
    events_list = []
    sharpe_list = []
    init_amt = amt

    #go through dates
    for date, r in tqdm (returns.iloc[14:].items(), total=len(returns.iloc[14:])):
        #if you're currently holding the stock, sell it
        

        #get data til just before current date
        curr_data = returns[:date]
        
        if type(order) == tuple:
            try:
                #fit model
                model = ARIMA(curr_data, order=order).fit(maxiter=200)

                #get forecast
                pred = model.forecast()[0][0]

            except:
                pred = thresh - 1


        #if you predict a high enough return and not holding, buy stock
        if ((type(order) == float and np.random.random() < order) 
         or (type(order) == tuple and pred > thresh)):
        
            
           
            buy_price = prices.loc[date]
            events_list.append(('b', date))
            int_buy_price = int(buy_price)
            sum_list.append(int_buy_price)
            if verbose:
                print('Bought at $%s'%buy_price)
                print('Predicted Return: %s'%round(pred,4))
                print('Actual Return: %s'%(round(ret, 4)))
                print('=======================================')
            continue

        #if you predict below the threshold return, sell the stock
        if ((type(order) == float and np.random.random() < order) 
         or (type(order) == tuple and thresh > pred)
         or (order == 'last' and curr_data[-1] > 0)):
            
            sell_price = prices.loc[date]
            
            total_return = len(sum_list) * sell_price 

            ret = (total_return-sum(sum_list))/sum(sum_list)
            amt *= (1+ret)
            events_list.append(('s', date, ret))
            sharpe_list.append(ret)
            sum_list.clear()

            if verbose:
                print('Sold at $%s'%sell_price)
                print('Predicted Return: %s'%round(pred,4))
                print('Actual Return: %s'%(round(ret, 4)))
                print('=======================================')
            

                
    if verbose:
        sharpe = sharpe_ratio(sharpe_list, risk_free_rate=0)
        print('Total Amount: $%s'%round(amt,2))
        print(f"Sharpe Ratio: {sharpe}")
        
    #graph
    if plot:
    
        plt.figure(figsize=(10,4))
        plt.plot(prices[14:])

        y_lims = (int(prices.min()*.95), int(prices.max()*1.05))
        shaded_y_lims = int(prices.min()*.5), int(prices.max()*1.5)

        for idx, event in enumerate(events_list):
            plt.axvline(event[1], color='k', linestyle='--', alpha=0.4)
            if event[0] == 's':
                color = 'green' if event[2] > 0 else 'red'
                plt.fill_betweenx(range(*shaded_y_lims), 
                                  event[1], events_list[idx-1][1], color=color, alpha=0.1)

        tot_return = round(100*(amt / init_amt - 1), 2)
        sharpe = sharpe_ratio(sharpe_list, risk_free_rate=0)
        tot_return = str(tot_return) + '%'
        plt.title("%s Price Data\nThresh=%s\nTotal Amt: $%s\nTotal Return: %s"%(tickerSymbol, thresh, round(amt,2), tot_return), fontsize=20)
        plt.ylim(*y_lims)
        plt.show()
        print(sharpe)
    
    return amt

# A model with a dth difference to fit and ARMA(p,q) model is called an ARIMA process 
# of order (p,d,q). You can select p,d, and q with a wide range of methods, 
# including AIC, BIC, and empirical autocorrelations (Petris, 2009).


for thresh in [0.001]:
    run_simulation(returns, prices, 100000, (7,1,7), thresh, verbose=True)

如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论, 你可以免费 注册 一个本站的账号。

评论(1

茶花眉 2025-01-16 20:09:14

我发现由于某种原因它无法拟合 ARIMA 模型。我想,没有收敛到解决方案。我不确定为什么,因为当我使用相同的数据和顺序运行不同的策略时,它非常适合。

I've discovered that it's failing to fit the ARIMA model for some reason. Not converging to a solution, I guess. I'm not sure why though because it was fitting it just fine when I was running a different strategy using the same data and order.

~没有更多了~
我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站,即表示您同意使用 Cookies 和您的相关数据。
原文