stock/futures_backtrader.py

import numpy as np
from sqlalchemy import create_engine
import pandas as pd
import pymysql
import backtrader as bt
import backtrader.indicators as btind
import datetime
import math
import threading
from datetime import datetime as dt
import multiprocessing as mp
from multiprocessing import Manager
from backtrader.feeds import PandasData
import os
# import multiprocessing
import matplotlib

class MyPandasData(PandasData):
    lines = ('change_pct', 'net_amount_main', 'net_pct_main', 'net_amount_xl', 'net_pct_xl', 'net_amount_l', 'net_pct_l'
             , 'net_amount_m', 'net_pct_m', 'net_amount_s', 'net_pct_s',)
    params = (('change_pct', 7),
              ('net_amount_main', 8),
              ('net_pct_main', 9),
              ('net_amount_xl', 10),
              ('net_pct_xl', 11),
              ('net_amount_l', 12),
              ('net_pct_l', 13),
              ('net_amount_m', 14),
              ('net_pct_m', 15),
              ('net_amount_s', 16),
              ('net_pct_s', 17),
              )


class TestStrategy(bt.Strategy):
    params = (
        ("num", 3),
        ('Volatility', 0),
        ('rate', 5),# 注意要有逗号！！
    )
    def log(self, txt, dt=None):
        ''' Logging function for this strategy'''
        dt = dt or self.datas[0].datetime.date(0)
        # print('%s, %s' % (dt.isoformat(), txt))

    def __init__(self):
        # self.num = num
        # self.Volatility = Volatility/100
        # Keep a reference to the "close" line in the data[0] dataseries
        self.dataclose = self.datas[0].close
        self.dataopen = self.datas[0].open
        self.high = self.datas[0].high
        self.low = self.datas[0].low
        self.volume = self.datas[0].volume
        self.change_pct = self.datas[0].change_pct
        self.net_amount_main = self.datas[0].net_amount_main
        self.net_pct_main = self.datas[0].net_pct_main
        self.net_amount_xl = self.datas[0].net_amount_xl
        self.net_pct_xl = self.datas[0].net_pct_xl
        self.net_amount_l = self.datas[0].net_amount_l
        self.net_pct_l = self.datas[0].net_pct_l
        self.sma5 = btind.MovingAverageSimple(self.datas[0].close, period=5)
        self.sma10 = btind.MovingAverageSimple(self.datas[0].close, period=10)
        self.sma20 = btind.MovingAverageSimple(self.datas[0].close, period=20)

    def notify_order(self, order):
        """
        订单状态处理

        Arguments:
            order {object} -- 订单状态
        """
        if order.status in [order.Submitted, order.Accepted]:
            # 如订单已被处理，则不用做任何事情
            return

        # 检查订单是否完成
        if order.status in [order.Completed]:
            if order.isbuy():
                self.buyprice = order.executed.price
                self.buycomm = order.executed.comm
            self.bar_executed = len(self)

        # 订单因为缺少资金之类的原因被拒绝执行
        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            pass
            # self.log('Order Canceled/Margin/Rejected')

        # 订单状态处理完成，设为空
        self.order = None

    def notify_trade(self, trade):
        """
        交易成果

        Arguments:
            trade {object} -- 交易状态
        """
        if not trade.isclosed:
            return

        # 显示交易的毛利率和净利润
        # self.log('OPERATION PROFIT, GROSS %.2f, NET %.2f' % (trade.pnl, trade.pnlcomm))

    def next(self):
        # print(self.num,self.Volatility)
        # Simply log the closing price of the series from the reference
        # self.sma20[-2] < self.sma20[-1] < self.sma20[0] and self.sma10[-2] < self.sma10[-1] < self.sma10[0]
        # and (self.sma5[-1] < self.sma10[-1])
        # and (self.net_pct_l[0] > 10) and (self.net_pct_xl[0] > 3)  \
        # and (self.net_amount_main[-1] > 0) and (self.net_amount_main[0] > 0)
        if len(self) > self.params.num:
            lowest = np.min(self.low.get(size=self.params.num))
            vola = self.params.Volatility/100
            rate = self.params.rate/100
            # print(f'{self.params.num}日天最低值：{lowest},波动率为{self.params.Volatility/100}')
            if (self.dataclose[0] > self.dataopen[0]) and ((lowest*(1-vola)) < self.low[0] < (lowest*(1+vola)))\
                    and (self.dataclose[0] > self.sma5[0]) and self.sma5[0] > self.sma5[-1] \
                    and (not self.position) and (self.sma5[0] > self.sma10[0]) \
                    and (self.net_pct_main[-2] > 5) \
                    and (self.change_pct[0] > 5):
                # self.log('BUY CREATE, %.2f' % self.dataclose[0])
                self.order = self.buy()
            elif self.dataclose < self.sma5[0] or self.sma5[0] < self.sma10[0]\
                    or (self.dataclose[0] > (self.sma5[0] * (1+rate))):
                self.order = self.close()
                # self.log('Close, %.2f' % self.dataclose[0])

    def stop(self):
        # pass
        self.log(u'(MA趋势交易效果) Ending Value %.2f' % (self.broker.getvalue()))



def backtrader(table_list, num, Volatility, rate, ns):
    stattime = dt.now()
    # print(type(ns.df), ns.df)
    engine = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/stocks_data?charset=utf8')
    result = mp.Manager().list()
    result_change = mp.Manager().list()
    result_change_fall = mp.Manager().list()
    err_list = mp.Manager().list()
    for stock in table_list:
        # print(stock)
        stk_df = pd.read_sql_table(stock, engine)
        stk_df.date = pd.to_datetime(stk_df.date)
        if len(stk_df) > 60:
            cerebro = bt.Cerebro()

            cerebro.addstrategy(TestStrategy, num=num, Volatility=Volatility, rate=rate)

            cerebro.addsizer(bt.sizers.FixedSize, stake=10000)
            data = MyPandasData(dataname=stk_df,
                                fromdate=datetime.datetime(2010,1,1),
                                todate=datetime.datetime(2022, 10, 30),
                                datetime='date',
                                open='open',
                                close='close',
                                high='high',
                                low='low',
                                volume='volume',
                                change_pct='change_pct',
                                net_amount_main='net_amount_main',
                                net_pct_main='net_pct_main',
                                net_amount_xl='net_amount_xl',
                                net_pct_xl='net_pct_xl',
                                net_amount_l='net_amount_l',
                                net_pct_l='net_pct_l',
                                net_amount_m='net_amount_m',
                                net_pct_m='net_pct_m',
                                net_amount_s='net_amount_s',
                                net_pct_s='net_pct_s',
                                )
            # print('取值完成')

            cerebro.adddata(data, name=stock)

            cerebro.broker.setcash(100000.0)
            cerebro.broker.setcommission(0.005)
            cerebro.addanalyzer(bt.analyzers.PyFolio)
            # 策略执行前的资金
            # print('启动资金: %.2f' % cerebro.broker.getvalue())
            try:
            # 策略执行
                cerebro.run()
            except IndexError:
                err_list.append(stock)
            else:
                if cerebro.broker.getvalue() > 100000.0:
                    # print('recode！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！')
                    result_change.append((cerebro.broker.getvalue()/10000-1))
                    result.append(stock)
                else:
                    result_change_fall.append((1-cerebro.broker.getvalue() / 10000))
                    # print('aaaaaaaaaaa')
    print(f'以{num}内最低值波动{Volatility}为支撑、{rate}%为乖离率，结果状态为：')
    print('正盈利的个股为：', len(result_change), '成功率为：', len(result)/len(table_list))
    print(f'总盈利:{np.sum(result_change)} 平均盈利:{np.mean(result_change)},最大盈利:{np.max(result_change)}, 最小盈利:{np.min(result_change)}')
    print(
        f'总亏损:{np.sum(result_change_fall)},平均亏损:{np.mean(result_change_fall)},最大亏损:{np.min(result_change_fall)} 最小亏损:{np.max(result_change_fall)}')
    endtime = dt.now()
    # ns.df.loc[len(ns.df)] = [num, Volatility, rate, len(result), len(result)/len(table_list), np.sum(result_change),
    #                                 np.mean(result_change), np.max(result_change), np.min(result_change),
    #                                 np.sum(result_change_fall), np.mean(result_change_fall),
    #                                 np.min(result_change_fall), np.max(result_change_fall)]
    a = pd.DataFrame(data=[[num, Volatility, rate, len(result), len(result)/len(table_list), np.sum(result_change),
                                    np.mean(result_change), np.max(result_change), np.min(result_change),
                                    np.sum(result_change_fall), np.mean(result_change_fall),
                                    np.min(result_change_fall), np.max(result_change_fall)]],
                     columns=['周期', '波动率', '乖离率', '盈利个数', '盈利比例', '总盈利', '平均盈利', '最大盈利', '最小盈利', '总亏损',
                               '平均亏损', '最大亏损', '最小亏损'])
    ns.df = pd.concat([ns.df, a], ignore_index=True)
    # ns.df = ns.df.append(a)
    print(ns.df)
    print('每轮耗时：', endtime-stattime)

            # cerebro.plot()

if __name__ == '__main__':
    # engine = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx?charset=utf8', poolclass=NullPool)

    # stocks = pd.read_sql_query(
    #                     'select value from MA5_1d', engine_hlfx)
    cpu_count = os.cpu_count()
    os_name = os.name
    if os_name == 'posix':
        path = '/Users/daniel/Documents/'
    else:
        path = 'D:\\BackTrader\\'
    fre = '1d'
    db = pymysql.connect(host='localhost',
                         user='root',
                         port=3307,
                         password='r6kEwqWU9!v3',
                         database='stocks_data')
    cursor = db.cursor()
    cursor.execute("show tables like '%%%s%%' " % fre)
    table_list = [tuple[0] for tuple in cursor.fetchall()]
    # print(table_list)
    # table_list = table_list[0:100]

    mgr = Manager()
    ns = mgr.Namespace()
    df = pd.DataFrame(columns=['周期', '波动率', '乖离率', '盈利个数', '盈利比例', '总盈利', '平均盈利', '最大盈利', '最小盈利', '总亏损',
                               '平均亏损', '最大亏损', '最小亏损'])
    ns.df = df
    sttime = dt.now()
    thread_list = []
    for num in range(30, 120, 10):
        for Volatility in range(3, 21, 1):
            for rate in range(5, 15, 1):
                #获得cpu数量，计算进程数
                # step = math.ceil(len(table_list)/cpu_count/100)*100
                print(f'{num}天波动率为{Volatility}% 乖离率为{rate}%')
                # for i in range(0, len(table_list), step):
                    # thd = threading.local()
                    # print(i)
                p = mp.Process(target=backtrader, args=(table_list,  num, Volatility, rate, ns))
                # p.start()
                thread_list.append(p)
                    # p.start()
                    # p.join()
    print(len(thread_list))
    print('开始轮循！')
    for thread in thread_list:
        thread.start()
    for thread in thread_list:
        thread.join()
    print('NS.DF:', '\n', ns.df)
    edtime = dt.now()
    print('总耗时：', edtime - sttime)
    ns.df.to_csv(path+dt.now().strftime('%Y-%m-%d')+'.csv', index=True,encoding='utf_8_sig')