back trader 回测更新

.idea/misc.xml

@@ -1,4 +1,4 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
-  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.8 (jqdata)" project-jdk-type="Python SDK" />
+  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.8" project-jdk-type="Python SDK" />
 </project>

.idea/modules.xml

@@ -6,6 +6,7 @@
       <module fileurl="file://$USER_HOME$/Library/CloudStorage/OneDrive-个人/个人/python_stocks/quantify01/.idea/quantify01.iml" filepath="$USER_HOME$/Library/CloudStorage/OneDrive-个人/个人/python_stocks/quantify01/.idea/quantify01.iml" />
       <module fileurl="file://$PROJECT_DIR$/.idea/stock.iml" filepath="$PROJECT_DIR$/.idea/stock.iml" />
       <module fileurl="file://$USER_HOME$/Library/CloudStorage/OneDrive-个人/个人/python_stocks/stocks_to_sql/.idea/stocks_to_sql.iml" filepath="$USER_HOME$/Library/CloudStorage/OneDrive-个人/个人/python_stocks/stocks_to_sql/.idea/stocks_to_sql.iml" />
+      <module fileurl="file://$PROJECT_DIR$/../../OneDrive/个人/python_stocks/xtQMT/.idea/xtQMT.iml" filepath="$PROJECT_DIR$/../../OneDrive/个人/python_stocks/xtQMT/.idea/xtQMT.iml" />
     </modules>
   </component>
 </project>

.idea/stock.iml

@@ -2,7 +2,7 @@
 <module type="PYTHON_MODULE" version="4">
   <component name="NewModuleRootManager">
     <content url="file://$MODULE_DIR$" />
-    <orderEntry type="jdk" jdkName="Python 3.8 (jqdata)" jdkType="Python SDK" />
+    <orderEntry type="jdk" jdkName="Python 3.8" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
     <orderEntry type="module" module-name="stocks_to_sql" />
     <orderEntry type="module" module-name="quantify01" />

futures_backtrader.py

@@ -0,0 +1,258 @@
+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 threading
+from datetime import datetime as dt
+import multiprocessing as mp
+from backtrader.feeds import PandasData
+
+# 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[-2] < (lowest*(1+vola))) or((lowest*(1-vola)) < self.low[-1] < (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, result, result_change,result_change_fall, num, Volatility, rate,err_list):
+    engine = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/stocks_data?charset=utf8')
+    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')
+
+
+            # 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)
+
+    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:500]
+
+    df = pd.DataFrame(columns=['周期', '波动率', '盈利个数', '盈利比例', '总盈利', '平均盈利', '最大盈利', '最小盈利', '总亏损',
+                               '平均亏损', '最大亏损', '最小亏损'])
+
+
+    sttime = dt.now()
+
+    for num in range(60, 140, 20):
+        for Volatility in range(8, 10, 1):
+            for rate in range(7, 10, 1):
+                step = 1000
+                thread_list = []
+                result = mp.Manager().list()
+                result_change = mp.Manager().list()
+                result_change_fall = mp.Manager().list()
+                err_list = mp.Manager().list()
+                print(f'{num}天波动率为{Volatility}%乖离率为{rate}')
+                for i in range(0, len(table_list), step):
+
+
+                    stattime = dt.now()
+                    # thd = threading.local()
+                    # print(i)
+                    p = mp.Process(target=backtrader, args=(table_list[i:i + step], result,
+                                                            result_change, result_change_fall, num, Volatility, rate,err_list))
+                    thread_list.append(p)
+                    # p.start()
+                    # p.join()
+                    # print(thread_list)
+                for thread in thread_list:
+                    thread.start()
+                for thread in thread_list:
+                    thread.join()
+
+                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()
+                df.loc[len(df)] = [num, Volatility, 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)]
+                print(df)
+                print('每轮耗时：', endtime-stattime)
+    edtime = dt.now()
+    print('总耗时：', edtime - sttime)
+    df.to_csv(r'C:\Users\Daniel\Documents\策略穷举2.csv', index=True)

real_time_signal.py

@@ -9,7 +9,7 @@ from datetime import datetime as dt
 
 start = dt.now()
 # 确定级别
-fre = '1d'
+fre = '30m'
 # 连接数据库
 db = pymysql.connect(host='localhost',
                      user='root',
@@ -24,7 +24,6 @@ cursor = db.cursor()
 cursor.execute("show tables like '%%%s%%' "% fre)
 # stocks = [tuple[0] for tuple in cursor.fetchall()]
 stocks = list(get_all_securities(['stock'], date='2021-12-31').index)
-stocks = ['301058.XSHE']
 # stocks = stocks[0:500]
 print(dt.now(), 'stocks范围已获取！')
 
@@ -124,19 +123,18 @@ def qbh_hlfx(stocks, df):
                             m = m-1
                     else:
                         thd.df_day.loc[x, 'HL'] = '-'
-                    print(thd.df_day)
             else:
                 pass
         except BaseException:
             continue
 
 
-# while True:
-df = get_bars(stocks, count=2, unit=fre,
-              fields=['date', 'open', 'close', 'high', 'low', 'volume', 'money'], include_now=True, df=True)
-print(dt.now(), 'get_bars 成功')
-# strattime = dt.now()
-qbh_hlfx(stocks, df)
+while True:
+    df = get_bars(stocks, count=2, unit=fre,
+                  fields=['date', 'open', 'close', 'high', 'low', 'volume', 'money'], include_now=True, df=True)
+    print(dt.now(), 'get_bars 成功')
+    # strattime = dt.now()
+    qbh_hlfx(stocks, df)
 # endtime = dt.now()
 
 

real_time_update_qbh_hlfx_1025.py

@@ -291,7 +291,7 @@ while True:
             stk = locals()
             thd = threading.local()
             # 进程准备
-            step = 600
+            step = 400
             thread_list = []
             engine_stock = []
             engine_hlfx = []

updata_qbh_hlfx.py

@@ -6,16 +6,24 @@ import threading
 from datetime import datetime as dt
 import datetime
 
-auth('18019403367', 'Qwer4321')
+# auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
 
-stocks = list(get_all_securities(['stock'], date=dt.today().strftime('%Y-%m-%d')).index)
+engine_stocks_list = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx_pool?charset=utf8')
+# stocks = list(get_all_securities(['stock'], date=dt.today().strftime('%Y-%m-%d')).index)
+
+
+stocks = pd.read_sql_query(
+    'select securities from stocks_list', engine_stocks_list)
+stocks = stocks.iloc[-1, 0]
+stocks = stocks.split(",")
+print(len(stocks), type(stocks), stocks)
 # stocks = stocks[0:1]
 
 start = dt.now()
 # 确定级别
 # 注意修改time delta
 # fre = '30m'
-for fre in ['30m', '1d']:
+for fre in ['1d', '30m']:
     start = dt.now()
     print(fre)
     # 连接数据库
@@ -38,97 +46,97 @@ for fre in ['30m', '1d']:
             if ('stk%s_%s' % (thd.stock, fre)) in table_list:
                 # 有历史数据
                 index_len = pd.read_sql_table('stk%s_%s' % (thd.stock, fre), con=engine2).iloc[-1, 0]
-                # 注意修改time delta
-                startdate = pd.read_sql_table('stk%s_%s' % (thd.stock, fre), con=engine2).iloc[-1, 1]
-                # startdate = pd.read_sql_table('stk%s_%s' % (stock, fre), con=engine2).iloc[-1, 1] + datetime.timedelta(minutes= 5)
-                thd.get_price = pd.read_sql_query(
-                    'select date,open,close,high,low,volume,money from `stk%s_%s`' % (thd.stock, fre), engine)
-                thd.get_price = thd.get_price.loc[thd.get_price['date'] > startdate]
-                thd.df_day = pd.read_sql_query(
-                    'select date,open,close,high,low,volume,money,HL from `stk%s_%s`' % (thd.stock, fre), engine2)
-                # 先处理去包含
-                for i in thd.get_price.index:
-                    # 不包含
-                    if (thd.df_day.iloc[-1, 3] > thd.get_price.loc[i, 'high']
-                        and thd.df_day.iloc[-1, 4] > thd.get_price.loc[i, 'low']) \
-                            or (thd.df_day.iloc[-1, 3] < thd.get_price.loc[i, 'high']
-                                and thd.df_day.iloc[-1, 4] < thd.get_price.loc[i, 'low']):
-                        thd.df_day = pd.concat([thd.df_day, thd.get_price.loc[[i]]], ignore_index=True)
-                        # print(thd.df_day)
-                    # 包含
-                    else:
-                        # (new_df.iloc[-1,3]>=df_day.loc[i,'high'] and new_df.iloc[-1,4]<= df_day.loc[i,'low']):
-                        # 左高，下降
-                        if thd.df_day.iloc[-2, 3] > thd.df_day.iloc[-1, 3]:
-                            thd.df_day.iloc[-1, 3] = min(thd.df_day.iloc[-1, 3], thd.get_price.loc[i, 'high'])
-                            thd.df_day.iloc[-1, 4] = min(thd.df_day.iloc[-1, 4], thd.get_price.loc[i, 'low'])
+                if index_len > 2:
+                    # 注意修改time delta
+                    startdate = pd.read_sql_table('stk%s_%s' % (thd.stock, fre), con=engine2).iloc[-1, 1]
+                    # startdate = pd.read_sql_table('stk%s_%s' % (stock, fre), con=engine2).iloc[-1, 1] + datetime.timedelta(minutes= 5)
+                    thd.get_price = pd.read_sql_query(
+                        'select date,open,close,high,low,volume,money from `stk%s_%s`' % (thd.stock, fre), engine)
+                    thd.get_price = thd.get_price.loc[thd.get_price['date'] > startdate]
+                    thd.df_day = pd.read_sql_query(
+                        'select date,open,close,high,low,volume,money,HL from `stk%s_%s`' % (thd.stock, fre), engine2)
+                    # 先处理去包含
+                    for i in thd.get_price.index:
+                        # 不包含
+                        if (thd.df_day.iloc[-1, 3] > thd.get_price.loc[i, 'high']
+                            and thd.df_day.iloc[-1, 4] > thd.get_price.loc[i, 'low']) \
+                                or (thd.df_day.iloc[-1, 3] < thd.get_price.loc[i, 'high']
+                                    and thd.df_day.iloc[-1, 4] < thd.get_price.loc[i, 'low']):
+                            thd.df_day = pd.concat([thd.df_day, thd.get_price.loc[[i]]], ignore_index=True)
+                            # print(thd.df_day)
+                        # 包含
                         else:
-                            # 右高，上升
-                            thd.df_day.iloc[-1, 3] = max(thd.df_day.iloc[-1, 3], thd.get_price.loc[i, 'high'])
-                            thd.df_day.iloc[-1, 4] = max(thd.df_day.iloc[-1, 4], thd.get_price.loc[i, 'low'])
-                            # 寻找顶底分型
-                if len(thd.df_day.index) > 2:
-                    # 寻找顶底分型
-                    for x in range(index_len, len(thd.df_day.index)):
-                        m = x - 1
-                        # 底
-                        if ((thd.df_day.loc[x, 'high'] > thd.df_day.loc[x - 1, 'high']) and (
-                                thd.df_day.loc[x - 2, 'high'] > thd.df_day.loc[x - 1, 'high'])):
-                            # if ((stk.df_day.loc[i-2, 'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-3,'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-1,'date'] != stk.fxdf.iloc[-1,0])):
-                            # stk.fxdf = pd.concat([stk.fxdf, stk.df_day.iloc[[i]]], ignore_index=True)
-                            thd.df_day.loc[x, 'HL'] = 'L*'
-                            while m:
-                                if thd.df_day.loc[m, 'HL'] == 'H':
-                                    if (x - m) > 3:
-                                        thd.df_day.loc[x, 'HL'] = 'L'
-                                        if x == len(thd.df_day.index) - 1:
-                                            print(thd.stock, '$$$$$$$', '\n', thd.df_day.loc[x, 'date'], '买买买买买！！')
-                                    break
-                                elif (thd.df_day.loc[m, 'HL'] == 'L'):
-                                    if thd.df_day.loc[x - 1, 'low'] < thd.df_day.loc[m - 1, 'low']:
-                                        # 前一个为底，且中间存在不包含 or 更低的底
+                            # (new_df.iloc[-1,3]>=df_day.loc[i,'high'] and new_df.iloc[-1,4]<= df_day.loc[i,'low']):
+                            # 左高，下降
+                            if thd.df_day.iloc[-2, 3] > thd.df_day.iloc[-1, 3]:
+                                thd.df_day.iloc[-1, 3] = min(thd.df_day.iloc[-1, 3], thd.get_price.loc[i, 'high'])
+                                thd.df_day.iloc[-1, 4] = min(thd.df_day.iloc[-1, 4], thd.get_price.loc[i, 'low'])
+                            else:
+                                # 右高，上升
+                                thd.df_day.iloc[-1, 3] = max(thd.df_day.iloc[-1, 3], thd.get_price.loc[i, 'high'])
+                                thd.df_day.iloc[-1, 4] = max(thd.df_day.iloc[-1, 4], thd.get_price.loc[i, 'low'])
+                                # 寻找顶底分型
+                    if len(thd.df_day.index) > 2:
+                        # 寻找顶底分型
+                        for x in range(index_len, len(thd.df_day.index)):
+                            m = x - 1
+                            # 底
+                            if ((thd.df_day.loc[x, 'high'] > thd.df_day.loc[x - 1, 'high']) and (
+                                    thd.df_day.loc[x - 2, 'high'] > thd.df_day.loc[x - 1, 'high'])):
+                                # if ((stk.df_day.loc[i-2, 'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-3,'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-1,'date'] != stk.fxdf.iloc[-1,0])):
+                                # stk.fxdf = pd.concat([stk.fxdf, stk.df_day.iloc[[i]]], ignore_index=True)
+                                thd.df_day.loc[x, 'HL'] = 'L*'
+                                while m:
+                                    if thd.df_day.loc[m, 'HL'] == 'H':
+                                        if (x - m) > 3:
+                                            thd.df_day.loc[x, 'HL'] = 'L'
+                                            if x == len(thd.df_day.index) - 1:
+                                                print(thd.stock, '$$$$$$$', '\n', thd.df_day.loc[x, 'date'], '买买买买买！！')
+                                        break
+                                    elif (thd.df_day.loc[m, 'HL'] == 'L'):
+                                        if thd.df_day.loc[x - 1, 'low'] < thd.df_day.loc[m - 1, 'low']:
+                                            # 前一个为底，且中间存在不包含 or 更低的底
+                                            thd.df_day.loc[x, 'HL'] = 'L'
+                                            if x == len(thd.df_day.index) - 1:
+                                                # pass
+                                                print(thd.stock, '$$$$$$$', '\n', thd.df_day.loc[x, 'date'], '中继后的底————买吗？！')
+                                            break
+                                        else:
+                                            break
+                                    m = m - 1
+                                    if m == 0:
                                         thd.df_day.loc[x, 'HL'] = 'L'
-                                        if x == len(thd.df_day.index) - 1:
-                                            # pass
-                                            print(thd.stock, '$$$$$$$', '\n', thd.df_day.loc[x, 'date'], '中继后的底————买吗？！')
+                            # 顶
+                            elif ((thd.df_day.loc[x, 'high'] < thd.df_day.loc[x - 1, 'high']) and (
+                                    thd.df_day.loc[x - 2, 'high'] < thd.df_day.loc[x - 1, 'high'])):
+                                # if ((stk.df_day.loc[i-2, 'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-3,'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-1,'date'] != stk.fxdf.iloc[-1,0])):
+                                #     stk.fxdf = pd.concat([stk.fxdf, stk.df_day.iloc[[i]]], ignore_index=True)
+                                thd.df_day.loc[x, 'HL'] = 'H*'
+                                while m:
+                                    if thd.df_day.loc[m, 'HL'] == 'L':
+                                        if x - m > 3:
+                                            thd.df_day.loc[x, 'HL'] = 'H'
+                                            if x == len(thd.df_day.index) - 1:
+                                                # print(stock, '!!!!!!!', '\n', '卖卖卖卖卖卖卖!')
+                                                pass
                                         break
-                                    else:
+                                    elif (thd.df_day.loc[m, 'HL'] == 'H'):
+                                        if thd.df_day.loc[x - 1, 'high'] > thd.df_day.loc[m - 1, 'high']:
+                                            # 前一个为顶，且中间存在不包含 or 更高的顶
+                                            thd.df_day.loc[x, 'HL'] = 'H'
+                                            if x == len(thd.df_day.index) - 1:
+                                                pass
+                                                # print(stock, '/\/\/\/\/\/\/', '一顶更有一顶高！')
+                                            break
                                         break
-                                m = m - 1
-                                if m == 0:
-                                    thd.df_day.loc[x, 'HL'] = 'L'
-                        # 顶
-                        elif ((thd.df_day.loc[x, 'high'] < thd.df_day.loc[x - 1, 'high']) and (
-                                thd.df_day.loc[x - 2, 'high'] < thd.df_day.loc[x - 1, 'high'])):
-                            # if ((stk.df_day.loc[i-2, 'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-3,'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-1,'date'] != stk.fxdf.iloc[-1,0])):
-                            #     stk.fxdf = pd.concat([stk.fxdf, stk.df_day.iloc[[i]]], ignore_index=True)
-                            thd.df_day.loc[x, 'HL'] = 'H*'
-                            while m:
-                                if thd.df_day.loc[m, 'HL'] == 'L':
-                                    if x - m > 3:
+                                    m = m - 1
+                                    if m == 0:
                                         thd.df_day.loc[x, 'HL'] = 'H'
-                                        if x == len(thd.df_day.index) - 1:
-                                            # print(stock, '!!!!!!!', '\n', '卖卖卖卖卖卖卖!')
-                                            pass
-                                        thd.df_day.loc[x, 9] = thd.df_day.loc[x, 'close'] - thd.df_day.loc[m, 'close']
-                                    break
-                                elif (thd.df_day.loc[m, 'HL'] == 'H'):
-                                    if thd.df_day.loc[x - 1, 'high'] > thd.df_day.loc[m - 1, 'high']:
-                                        # 前一个为顶，且中间存在不包含 or 更高的顶
-                                        thd.df_day.loc[x, 'HL'] = 'H'
-                                        if x == len(thd.df_day.index) - 1:
-                                            pass
-                                            # print(stock, '/\/\/\/\/\/\/', '一顶更有一顶高！')
-                                        break
-                                    break
-                                m = m - 1
-                                if m == 0:
-                                    thd.df_day.loc[x, 'HL'] = 'H'
-                        else:
-                            thd.df_day.loc[x, 'HL'] = '-'
+                            else:
+                                thd.df_day.loc[x, 'HL'] = '-'
 
-                # 更新数据库
-                thd.df_day[index_len + 1:].to_sql('stk%s_%s' % (thd.stock, fre), con=engine2, index=True, if_exists='append')
+                    # 更新数据库
+                    thd.df_day[index_len + 1:].to_sql('stk%s_%s' % (thd.stock, fre), con=engine2, index=True, if_exists='append')
             else:
                 # 没有历史数据表
                 thd.df_day = pd.DataFrame(columns=('date', 'open', 'close', 'high', 'low', 'volume', 'money', 'HL'))
@@ -197,7 +205,6 @@ for fre in ['30m', '1d']:
                                         if x == len(thd.df_day.index) - 1:
                                             # print(stock, '!!!!!!!', '\n', '卖卖卖卖卖卖卖!')
                                             pass
-                                        thd.df_day.loc[x, 9] = thd.df_day.loc[x, 'close'] - thd.df_day.loc[m, 'close']
                                     break
                                 elif (thd.df_day.loc[m, 'HL'] == 'H'):
                                     if thd.df_day.loc[x - 1, 'high'] > thd.df_day.loc[m - 1, 'high']:
@@ -216,11 +223,12 @@ for fre in ['30m', '1d']:
                 # 更新数据库
                 thd.df_day.to_sql('stk%s_%s' % (thd.stock, fre), con=engine2, index=True, if_exists='append')
 
-    step = 500
+    step = 700
     thread_list = []
     engine = []
     engine2 = []
     times_engine = 0
+    print(len(stocks))
     for i in range(0, len(stocks), step):
         engine.append(create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/stocks?charset=utf8'))
         engine2.append(create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx?charset=utf8'))

updata_qbh_hlfx_1114.py

@@ -0,0 +1,250 @@
+import multiprocessing as mp
+import pandas as pd
+import pymysql
+from sqlalchemy import create_engine
+from datetime import datetime as dt
+
+
+import datetime
+
+# auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
+def hlfx(stocks,fre,table_list):
+    engine = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/stocks?charset=utf8')
+    engine2 = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx?charset=utf8')
+    for stock in stocks:
+        # print(stock)
+        if ('stk%s_%s' % (stock, fre)) in table_list:
+            # 有历史数据
+            index_len = pd.read_sql_table('stk%s_%s' % (stock, fre), con=engine2).iloc[-1, 0]
+
+            # 注意修改time delta
+            startdate = pd.read_sql_table('stk%s_%s' % (stock, fre), con=engine2).iloc[-1, 1]
+            # startdate = pd.read_sql_table('stk%s_%s' % (stock, fre), con=engine2).iloc[-1, 1] + datetime.timedelta(minutes= 5)
+            get_price = pd.read_sql_query(
+                'select date,open,close,high,low,volume,money from `stk%s_%s`' % (stock, fre), engine)
+            get_price = get_price.loc[get_price['date'] > startdate]
+            df_day = pd.read_sql_query(
+                'select date,open,close,high,low,volume,money,HL from `stk%s_%s`' % (stock, fre), engine2)
+            if index_len > 2:
+                # 先处理去包含
+                for i in get_price.index:
+                    # 不包含
+                    if (df_day.iloc[-1, 3] > get_price.loc[i, 'high']
+                        and df_day.iloc[-1, 4] > get_price.loc[i, 'low']) \
+                            or (df_day.iloc[-1, 3] < get_price.loc[i, 'high']
+                                and df_day.iloc[-1, 4] < get_price.loc[i, 'low']):
+                        df_day = pd.concat([df_day, get_price.loc[[i]]], ignore_index=True)
+                        # print(df_day)
+                    # 包含
+                    else:
+                        # (new_df.iloc[-1,3]>=df_day.loc[i,'high'] and new_df.iloc[-1,4]<= df_day.loc[i,'low']):
+                        # 左高，下降
+                        if df_day.iloc[-2, 3] > df_day.iloc[-1, 3]:
+                            df_day.iloc[-1, 3] = min(df_day.iloc[-1, 3], get_price.loc[i, 'high'])
+                            df_day.iloc[-1, 4] = min(df_day.iloc[-1, 4], get_price.loc[i, 'low'])
+                        else:
+                            # 右高，上升
+                            df_day.iloc[-1, 3] = max(df_day.iloc[-1, 3], get_price.loc[i, 'high'])
+                            df_day.iloc[-1, 4] = max(df_day.iloc[-1, 4], get_price.loc[i, 'low'])
+                            # 寻找顶底分型
+                if len(df_day.index) > 2:
+                    # 寻找顶底分型
+                    for x in range(index_len, len(df_day.index)):
+                        m = x - 1
+                        # 底
+                        if ((df_day.loc[x, 'high'] > df_day.loc[x - 1, 'high']) and (
+                                df_day.loc[x - 2, 'high'] > df_day.loc[x - 1, 'high'])):
+                            # if ((stk.df_day.loc[i-2, 'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-3,'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-1,'date'] != stk.fxdf.iloc[-1,0])):
+                            # stk.fxdf = pd.concat([stk.fxdf, stk.df_day.iloc[[i]]], ignore_index=True)
+                            df_day.loc[x, 'HL'] = 'L*'
+                            while m:
+                                if df_day.loc[m, 'HL'] == 'H':
+                                    if (x - m) > 3:
+                                        df_day.loc[x, 'HL'] = 'L'
+                                        if x == len(df_day.index) - 1:
+                                            # print(stock, '$$$$$$$', '\n', df_day.loc[x, 'date'], '买买买买买！！')
+                                            pass
+                                    break
+                                elif (df_day.loc[m, 'HL'] == 'L'):
+                                    if df_day.loc[x - 1, 'low'] < df_day.loc[m - 1, 'low']:
+                                        # 前一个为底，且中间存在不包含 or 更低的底
+                                        df_day.loc[x, 'HL'] = 'L'
+                                        if x == len(df_day.index) - 1:
+                                            pass
+                                            # print(stock, '$$$$$$$', '\n', df_day.loc[x, 'date'],
+                                            #       '中继后的底————买吗？！')
+                                        break
+                                    else:
+                                        break
+                                m = m - 1
+                                if m == 0:
+                                    df_day.loc[x, 'HL'] = 'L'
+                        # 顶
+                        elif ((df_day.loc[x, 'high'] < df_day.loc[x - 1, 'high']) and (
+                                df_day.loc[x - 2, 'high'] < df_day.loc[x - 1, 'high'])):
+                            # if ((stk.df_day.loc[i-2, 'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-3,'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-1,'date'] != stk.fxdf.iloc[-1,0])):
+                            #     stk.fxdf = pd.concat([stk.fxdf, stk.df_day.iloc[[i]]], ignore_index=True)
+                            df_day.loc[x, 'HL'] = 'H*'
+                            while m:
+                                if df_day.loc[m, 'HL'] == 'L':
+                                    if x - m > 3:
+                                        df_day.loc[x, 'HL'] = 'H'
+                                        if x == len(df_day.index) - 1:
+                                            # print(stock, '!!!!!!!', '\n', '卖卖卖卖卖卖卖!')
+                                            pass
+                                    break
+                                elif (df_day.loc[m, 'HL'] == 'H'):
+                                    if df_day.loc[x - 1, 'high'] > df_day.loc[m - 1, 'high']:
+                                        # 前一个为顶，且中间存在不包含 or 更高的顶
+                                        df_day.loc[x, 'HL'] = 'H'
+                                        if x == len(df_day.index) - 1:
+                                            pass
+                                            # print(stock, '/\/\/\/\/\/\/', '一顶更有一顶高！')
+                                        break
+                                    break
+                                m = m - 1
+                                if m == 0:
+                                    df_day.loc[x, 'HL'] = 'H'
+                        else:
+                            df_day.loc[x, 'HL'] = '-'
+                # 更新数据库
+                df_day[index_len + 1:].to_sql('stk%s_%s' % (stock, fre), con=engine2, index=True,
+                                              if_exists='append')
+            else:
+                df_day = pd.concat([df_day, get_price], ignore_index=True)
+                df_day[index_len + 1:].to_sql('stk%s_%s' % (stock, fre), con=engine2, index=True,
+                                              if_exists='append')
+        else:
+            # 没有历史数据表
+            df_day = pd.DataFrame(columns=('date', 'open', 'close', 'high', 'low', 'volume', 'money', 'HL'))
+            get_price = pd.read_sql_query(
+                'select date,open,close,high,low,volume,money from `stk%s_%s`' % (stock, fre), engine)
+            # 先处理去包含
+            for i in get_price.index:
+                if i == 0 or i == 1:
+                    df_day = pd.concat([df_day, get_price.iloc[[i]]], ignore_index=True)
+                # 不包含
+                elif (df_day.iloc[-1, 3] > get_price.loc[i, 'high']
+                      and df_day.iloc[-1, 4] > get_price.loc[i, 'low']) \
+                        or (df_day.iloc[-1, 3] < get_price.loc[i, 'high']
+                            and df_day.iloc[-1, 4] < get_price.loc[i, 'low']):
+                    df_day = pd.concat([df_day, get_price.loc[[i]]], ignore_index=True)
+                # 包含
+                else:
+                    # 左高，下降
+                    if df_day.iloc[-2, 3] > df_day.iloc[-1, 3]:
+                        df_day.iloc[-1, 3] = min(df_day.iloc[-1, 3], get_price.loc[i, 'high'])
+                        df_day.iloc[-1, 4] = min(df_day.iloc[-1, 4], get_price.loc[i, 'low'])
+                    else:
+                        # 右高，上升
+                        df_day.iloc[-1, 3] = max(df_day.iloc[-1, 3], get_price.loc[i, 'high'])
+                        df_day.iloc[-1, 4] = max(df_day.iloc[-1, 4], get_price.loc[i, 'low'])
+            if len(df_day.index) > 2:
+                # 寻找顶底分型
+                for x in range(2, len(df_day.index)):
+                    m = x - 1
+                    # 底
+                    if ((df_day.loc[x, 'high'] > df_day.loc[x - 1, 'high']) and (
+                            df_day.loc[x - 2, 'high'] > df_day.loc[x - 1, 'high'])):
+                        # if ((stk.df_day.loc[i-2, 'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-3,'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-1,'date'] != stk.fxdf.iloc[-1,0])):
+                        # stk.fxdf = pd.concat([stk.fxdf, stk.df_day.iloc[[i]]], ignore_index=True)
+                        df_day.loc[x, 'HL'] = 'L*'
+                        while m:
+                            if df_day.loc[m, 'HL'] == 'H':
+                                if (x - m) > 3:
+                                    df_day.loc[x, 'HL'] = 'L'
+                                    if x == len(df_day.index) - 1:
+                                        pass
+                                        # print(stock, '$$$$$$$', '\n', df_day.loc[x, 'date'], '买买买买买！！')
+                                break
+                            elif (df_day.loc[m, 'HL'] == 'L'):
+                                if df_day.loc[x - 1, 'low'] < df_day.loc[m - 1, 'low']:
+                                    # 前一个为底，且中间存在不包含 or 更低的底
+                                    df_day.loc[x, 'HL'] = 'L'
+                                    if x == len(df_day.index) - 1:
+                                        pass
+                                        # print(stock, '$$$$$$$', '\n', df_day.loc[x, 'date'], '中继后的底————买吗？！')
+                                    break
+                                else:
+                                    break
+                            m = m - 1
+                            if m == 0:
+                                df_day.loc[x, 'HL'] = 'L'
+                    # 顶
+                    elif ((df_day.loc[x, 'high'] < df_day.loc[x - 1, 'high']) and (
+                            df_day.loc[x - 2, 'high'] < df_day.loc[x - 1, 'high'])):
+                        # if ((stk.df_day.loc[i-2, 'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-3,'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-1,'date'] != stk.fxdf.iloc[-1,0])):
+                        #     stk.fxdf = pd.concat([stk.fxdf, stk.df_day.iloc[[i]]], ignore_index=True)
+                        df_day.loc[x, 'HL'] = 'H*'
+                        while m:
+                            if df_day.loc[m, 'HL'] == 'L':
+                                if x - m > 3:
+                                    df_day.loc[x, 'HL'] = 'H'
+                                    if x == len(df_day.index) - 1:
+                                        # print(stock, '!!!!!!!', '\n', '卖卖卖卖卖卖卖!')
+                                        pass
+                                break
+                            elif (df_day.loc[m, 'HL'] == 'H'):
+                                if df_day.loc[x - 1, 'high'] > df_day.loc[m - 1, 'high']:
+                                    # 前一个为顶，且中间存在不包含 or 更高的顶
+                                    df_day.loc[x, 'HL'] = 'H'
+                                    if x == len(df_day.index) - 1:
+                                        pass
+                                        # print(stock, '/\/\/\/\/\/\/', '一顶更有一顶高！')
+                                    break
+                                break
+                            m = m - 1
+                            if m == 0:
+                                df_day.loc[x, 'HL'] = 'H'
+                    else:
+                        df_day.loc[x, 'HL'] = '-'
+            # 更新数据库
+            df_day.to_sql('stk%s_%s' % (stock, fre), con=engine2, index=True, if_exists='append')
+
+
+if __name__ == '__main__':
+    engine_stocks_list = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx_pool?charset=utf8')
+    # stocks = list(get_all_securities(['stock'], date=dt.today().strftime('%Y-%m-%d')).index)
+
+
+    stocks = pd.read_sql_query(
+        'select securities from stocks_list', engine_stocks_list)
+    stocks = stocks.iloc[-1, 0]
+    stocks = stocks.split(",")
+    print(len(stocks), type(stocks), stocks)
+    # stocks = stocks[0:1000]
+
+    start = dt.now()
+    # 确定级别
+    # 注意修改time delta
+    # fre = '30m'
+
+    for fre in ['30m', '1d']:
+        start = dt.now()
+        print(fre)
+        # 连接数据库
+        db = pymysql.connect(host='localhost',
+                             user='root',
+                             port=3307,
+                             password='r6kEwqWU9!v3',
+                             database='hlfx')
+        cursor = db.cursor()
+        cursor.execute("show tables like '%%%s%%' " % fre)
+        table_list = [tuple[0] for tuple in cursor.fetchall()]
+        print('取得 table_list %s' % fre)
+
+        step = 800
+        mp_list = []
+        print(len(stocks))
+
+        for i in range(0, len(stocks), step):
+            p = mp.Process(target=hlfx, args=(stocks[i:i + step], fre, table_list, ))
+            mp_list.append(p)
+            p.start()
+
+        for processing in mp_list:
+            processing.join()
+        # db.close()
+
+        end = dt.now()
+        print('总时长：', (end - start).seconds)

update_data_tosql.py

@@ -3,15 +3,29 @@ from sqlalchemy import create_engine
 import pandas as pd
 from datetime import datetime as dt
 import datetime
+import pymysql
 
 auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
 stocks = list(get_all_securities(['stock'], date=dt.today().strftime('%Y-%m-%d')).index)
 engine = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/stocks?charset=utf8')
+stocks_List = ','.join(set(stocks))
+
+db_stocks_list = pymysql.connect(host='localhost',
+                          user='root',
+                          port=3307,
+                          password='r6kEwqWU9!v3',
+                          database='hlfx_pool')
+cursor_stock_list = db_stocks_list.cursor()
+sql = "INSERT INTO stocks_list (date,securities) VALUES('%s','%s')" % (dt.today().strftime('%Y-%m-%d'),  stocks_List)
+cursor_stock_list.execute(sql)
+db_stocks_list.commit()
+db_stocks_list.close()
+print('stocks_list已入库')
 
 for fre in ['30m', '1d']:
     print('ready to write to mysql %s' % fre)
     for stock in stocks:
-        print(stock)
+        print(stock, fre)
         try:
             index_len = pd.read_sql_table('stk%s_%s' % (stock, fre), con=engine).iloc[-1, 0]
             # 注意修改time delta
@@ -32,7 +46,8 @@ for fre in ['30m', '1d']:
             df_stock.index = df_stock.index + index_len + 1
             df_stock.to_sql('stk%s_%s' % (stock, fre), con=engine, index=True, if_exists='append')
         except BaseException:
-            df_stock = get_price(stock, start_date='2022-01-01 00:00:00', end_date=dt.today().strftime('%Y-%m-%d %H:%M:%S'),
+            df_stock = get_price(stock, start_date='2022-01-01 00:00:00',
+                                 end_date=dt.today().strftime('%Y-%m-%d %H:%M:%S'),
                                  frequency=fre, fields=['open', 'close', 'high', 'low', 'volume', 'money'],
                                  skip_paused=False,
                                  fq='pre', count=None, panel=False)