基于皮尔逊系数实现股票预测【多线程】

# -*- coding: utf-8 -*-
"""
Created on Tue Dec  4 08:53:08 2018

@author: zhen
"""
from dtw import fastdtw
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import threading
import time
from datetime import datetime

def normalization(x): # np.std:计算矩阵的标准差（方差的算术平方根）
    return (x - np.mean(x)) / np.std(x)

def corrcoef(a,b):
    corrc = np.corrcoef(a,b) # 计算皮尔逊相关系数，用于度量两个变量之间的相关性，其值介于-1到1之间
    corrc = corrc[0,1]
    return (16 * ((1 - corrc) / (1 + corrc)) ** 1) # ** 表示乘方
        
startTimeStamp = datetime.now() # 获取当前时间
# 加载数据
filename = 'C:/Users/zhen/.spyder-py3/sh000300_2017.csv'
# 获取第一，二列的数据
all_date = pd.read_csv(filename,usecols=[0, 1], dtype = 'str')
all_date = np.array(all_date)
data = all_date[:, 0]
times = all_date[:, 1]

data_points = pd.read_csv(filename,usecols=[3])
data_points = np.array(data_points)
data_points = data_points[:,0] #数据

topk = 10 #只显示top-10
baselen = 100 # 假设在50到150之间变化
basebegin = 365
basedata = data[basebegin]+' '+times[basebegin]+'~'+data[basebegin+baselen-1]+' '+times[basebegin+baselen-1]
length = len(data_points) #数据长度

# 定义自定义线程类
class Thread_Local(threading.Thread):
    def __init__(self, thread_id, name, counter):
        threading.Thread.__init__(self)
        self.thread_id = thread_id
        self.name = name
        self.counter = counter
        self.__running = threading.Event() # 标识停止线程
        self.__running.set() # 设置为True
        
    def run(self):
        print("starting %s" % self.name)
        split_data(self, self.counter) # 执行代码逻辑
        
    def stop(self):
        self.__running.clear()
        
# 分割片段并执行匹配，多线程
def split_data(self, split_len):
    base = data_points[basebegin:basebegin+split_len]  # 获取初始要匹配的数据
    subseries = []
    dateseries = []
    for j in range(0, length): 
        if (j < (basebegin - split_len) or j > (basebegin + split_len - 1)) and j <length - split_len:
            subseries.append(data_points[j:j+split_len])
            dateseries.append(j) #开始位置
    search(self, subseries, base, dateseries)  # 调用模式匹配

# 定义结果变量
result = []  
base_list = []
date_list = []
def search(self, subseries, base, dateseries):
     # 片段搜索
    listdistance = []
    for i in range(0, len(subseries)):
        tt = np.array(subseries[i])
        # dist, cost, acc, path = fastdtw(base, tt, dist='euclidean')
        # listdistance.append(dist)
        distance = corrcoef(base, tt)
        listdistance.append(distance)
    # 排序
    index = np.argsort(listdistance, kind='quicksort') #排序，返回排序后的索引序列
    result.append(subseries[index[0]])
    print("result length is %d" % len(result))
    base_list.append(base)
    date_list.append(dateseries[index[0]])
    # 关闭线程
    self.stop()
    
# 变换数据（收缩或扩展），生成50到150之间的数据，间隔为10
loc = 0
for split_len in range(round(0.5 * baselen), round(1.5 * baselen), 10):
    # 执行匹配
   thread = Thread_Local(1, "Thread" + str(loc), split_len)
   loc += 1
   # 开启线程
   thread.start()

boo = 1

while(boo > 0):
    if(len(result) < 10):
        if(boo % 100 == 0):
            print("has running %d s" % boo)
        boo += 1
        time.sleep(1)
    else:
        boo = 0
        
 # 片段搜索
listdistance = []
for i in range(0, len(result)):
    tt = np.array(result[i])
    distance = corrcoef(base_list[i], tt)
    listdistance.append(distance)
# 最终排序   
index = np.argsort(listdistance, kind='quicksort') #排序，返回排序后的索引序列
print("closed Main Thread")
endTimeStamp = datetime.now()
# 结果集对比
plt.figure(0)
plt.plot(normalization(base_list[index[0]]),label= basedata,linewidth='2')
length = len(result[index[0]])
begin = data[date_list[index[0]]] + ' ' + times[date_list[index[0]]]
end = data[date_list[index[0]] + length - 1] + ' ' + times[date_list[index[0]] + length - 1]
label = begin + '~' + end
plt.plot(normalization(result[index[0]]), label=label, linewidth='2')  
plt.legend(loc='upper left')
plt.title('normal similarity search')
plt.show()
print('run time', (endTimeStamp-startTimeStamp).seconds, "s")

结果：

分析：

　　皮尔逊相关系数（corrcoef）运算速度远超DTW或FASTDTW，但DTW或FASTDTW应用范围更广，适用于等长或变长的比较。