การทำนายราคาหุ้นโดยใช้ Hidden Markov Model | รหัสของรูบิค (2023)

กโมเดลมาร์คอฟที่ซ่อนอยู่(อืม) เป็นกรณีเฉพาะของโมเดลพื้นที่สถานะซึ่งตัวแปรแฝงเป็นตัวแปรแยกและตัวแปรพหุนาม จากการแสดงกราฟิก คุณสามารถถือว่า HMM เป็นกระบวนการสุ่มสองครั้งซึ่งประกอบด้วยกระบวนการมาร์กอฟสุ่มที่ซ่อนอยู่ (ของตัวแปรแฝง) ที่คุณไม่สามารถสังเกตได้โดยตรง และกระบวนการสุ่มอีกแบบหนึ่งที่สร้างลำดับของการสังเกตตามกระบวนการแรก

HMM มีความสามารถในการทำนายและวิเคราะห์ปรากฏการณ์ตามเวลา ดังนั้น จึงมีประโยชน์อย่างมากในด้านต่างๆ เช่น การรู้จำเสียง การประมวลผลภาษาธรรมชาติ และการทำนายตลาดการเงิน ในบทความนี้ คุณจะดูการประยุกต์ใช้ HMM ในด้านการวิเคราะห์ตลาดการเงิน ซึ่งส่วนใหญ่เป็นการทำนายราคาหุ้น

ในบทความนี้เราครอบคลุม:

1. การทำนายราคาหุ้น
2. การรวบรวมข้อมูลราคาหุ้น
3. คุณสมบัติสำหรับการทำนายราคาหุ้น
4. การทำนายราคาโดยใช้ HMM

การทำนายตลาดหุ้นเป็นหนึ่งในการวิจัยที่มีการใช้งานมากขึ้นในอดีต เนื่องจากได้รับความสนใจอย่างชัดเจนจากบริษัทยักษ์ใหญ่หลายแห่ง ในอดีต อัลกอริธึมการเรียนรู้ของเครื่องต่างๆ ได้ถูกนำมาใช้กับระดับความสำเร็จที่แตกต่างกันไป

อย่างไรก็ตาม การพยากรณ์หุ้นยังคงมีข้อจำกัดอย่างมากเนื่องจากธรรมชาติไม่คงที่ เป็นไปตามฤดูกาล และไม่สามารถคาดเดาได้ การทำนายการพยากรณ์จากข้อมูลหุ้นก่อนหน้าเป็นงานที่ท้าทายยิ่งกว่า เนื่องจากเป็นการละเลยปัจจัยภายนอกหลายประการ

HMM สามารถสร้างแบบจำลองการเปลี่ยนสถานะที่ซ่อนอยู่จากข้อมูลที่สังเกตตามลำดับ ปัญหาของการทำนายหุ้นสามารถคิดตามรูปแบบเดียวกันได้ ราคาของหุ้นขึ้นอยู่กับปัจจัยหลายอย่าง ซึ่งโดยทั่วไปแล้วนักลงทุนจะมองไม่เห็น (ตัวแปรที่ซ่อนอยู่)

การเปลี่ยนแปลงระหว่างปัจจัยพื้นฐานจะเปลี่ยนแปลงไปตามนโยบายและการตัดสินใจของบริษัท สภาวะทางการเงินของบริษัท และการตัดสินใจของฝ่ายบริหาร ซึ่งสิ่งเหล่านี้ส่งผลต่อราคาของหุ้น (ข้อมูลที่สังเกตได้) ดังนั้น HMM จึงเหมาะสมอย่างยิ่งกับปัญหาการคาดการณ์ราคา

ตอนนี้คุณสามารถทดสอบได้โดยการคาดการณ์ราคาหุ้นสำหรับ Alphabet Inc. (GOOGL), Facebook (FB) และ Apple Inc. (AAPL) ด้วย HMM

ใช้ไพสต็อคข้อมูล(http://data.pystock.com) เพื่อรับข้อมูลราคาหุ้นย้อนหลัง ทุกวัน ก่อนที่ตลาดหุ้นสหรัฐจะเปิดทำการเวลา 9:30 น. EST/EDTไพสต็อคโปรแกรมรวบรวมข้อมูลรวบรวมราคาหุ้นและรายงานทางการเงิน และส่งข้อมูล เช่น ราคาเปิด ราคาปิด ราคาสูงสุด และราคาต่ำสุดของหุ้นไปยังที่เก็บ ข้อมูลนี้อิงตามวัน ซึ่งหมายความว่าจะไม่มีข้อมูลระดับชั่วโมงหรือนาที

ดาวน์โหลดไพสต็อคข้อมูลสำหรับปีที่กำหนด เนื่องจากชุดข้อมูลมีขนาดใหญ่ ให้สร้างสคริปต์ Python เพื่อดาวน์โหลดข้อมูลในปีที่กำหนด และรันโปรแกรมพร้อมกันเป็นเวลาสามปีเพื่อดาวน์โหลดข้อมูลทั้งหมดพร้อมกัน:

"""การใช้งาน: get_data.py --year="""คำขอนำเข้านำเข้า osfrom docopt นำเข้า docopt # docopt ช่วยแยกวิเคราะห์อาร์กิวเมนต์บรรทัดคำสั่งใน # ในลักษณะง่ายๆ (http://docopt.org/)args = docopt (doc=__doc__, argv=None,help=True, version=None,options_first=False) year = args['--year'] # สร้างไดเร็กทอรีหากไม่ใช่ presentyear_directory_name = 'data/{year}'.format(year= year)ถ้าไม่ใช่ os.path.exists(year_directory_name): os.makedirs(year_directory_name) # กำลังดึงรายการไฟล์สำหรับ yearyear_data_files ที่สอดคล้องกัน = request.get('http://data.pystock.com/{year}/index.txt '.format(year=year)).text.strip().split('\n') สำหรับ data_file_name ใน year_data_files: file_location = '{year_directory_name}/{data_file_name}'.format(year_directory_name=year_directory_name,data_file_name=data_file_name) ด้วย open(file_location, 'wb+') เป็น data_file:print('>>> กำลังดาวน์โหลด \t {file_location}'.format(file_location=file_location)) data_file_content = request.get('http://data.pystock.com/ {year}/{data_file_name}'.format(year=year, data_file_name=data_file_name) ).contentprint('<<< ดาวน์โหลดเสร็จ \t {file_location}'.format(file_location=file_location)) data_file.write(data_file_content)

python get_data.py -- ปี 2015python get_data.py -- ปี 2016python get_data.py -- ปี 2017

"""การใช้งาน: parse_data.py --company="""นำเข้า osimport tarfileimport pandas เป็น pdfrom pandas นำเข้าข้อผิดพลาดเป็น pd_errorsfrom functools นำเข้า reducefrom docopt นำเข้า docopt args = docopt(doc=__doc__, argv=None,help=True , version=None,options_first=False) years = [2015, 2016, 2017]company = args['--company'] # รับไฟล์ข้อมูล listdata_files_list = []สำหรับปีในปี: year_directory = 'data/{year} '.format(year=year)สำหรับไฟล์ใน os.listdir(year_directory): data_files_list.append('{year_directory}/{file}'.format(year_directory=year_directory, file=file)) def parse_data(file_name, company_ symbol) :""" ส่งคืนข้อมูลสำหรับบริษัทที่เกี่ยวข้อง :param file_name: ชื่อของไฟล์ tar:param company_ symbol: สัญลักษณ์บริษัท:ประเภท file_name: str:ประเภท company_ symbol: str:return: dataframe สำหรับข้อมูลบริษัทที่เกี่ยวข้อง:rtype: pd.DataFrame """tar = tarfile.open(file_name)ลอง: price_report = pd.read_csv(tar.extractfile('prices.csv')) company_price_data = price_report[price_report[' symbol'] == company_ symbol]return company_price_dataexcept (KeyError, pd_errors .EmptyDataError):ส่งคืน pd.DataFrame() # รับข้อมูลที่สมบูรณ์สำหรับบริษัทที่กำหนดcompany_data = ลด (แลมบ์ดา df, file_name: df.append(parse_data(ชื่อไฟล์, บริษัท)),data_files_list,pd.DataFrame())company_data = company_data .sort_values(by=['date']) # สร้างโฟลเดอร์สำหรับข้อมูลบริษัทหากไม่มีอยู่หากไม่มี os.path.exists('data/company_data'): os.makedirs('data/company_data') # เขียนข้อมูลไปยัง ไฟล์ CSVcompany_data.to_csv('data/company_data/{company}.csv'.format(company=company),columns=['date', 'open', 'high', 'low', 'close', 'volume' , 'adj_close'],ดัชนี=เท็จ)

python parse_data.py --บริษัท GOOGLpython parse_data.py --บริษัท FBpython parse_data.py --บริษัท AAPL

คุณมีคุณสมบัติที่จำกัดมากในแต่ละวัน ได้แก่ ราคาเปิดของหุ้นในวันนั้น ราคาปิด ราคาสูงสุดของหุ้น และราคาต่ำสุดของหุ้น ดังนั้นใช้มันเพื่อคำนวณราคาหุ้น คุณสามารถคำนวณราคาหุ้นปิดของวัน โดยพิจารณาจากราคาหุ้นเปิดของวันนั้นและก่อนหน้างข้อมูลวัน ตัวทำนายของคุณจะมีเวลาแฝงของงวัน

ตอนนี้สร้างตัวทำนายที่เรียกว่าทำนายหุ้นซึ่งจะมีตรรกะทั้งหมดในการทำนายราคาหุ้นของบริษัทหนึ่งๆ ในแต่ละวัน

แทนที่จะใช้ราคาเปิด ปิด ราคาต่ำ และราคาสูงของหุ้นโดยตรง ให้แยกการเปลี่ยนแปลงเศษส่วนในแต่ละรายการที่จะใช้ในการฝึก HMM ของคุณ กำหนดพารามิเตอร์เหล่านี้ดังต่อไปนี้:

สำหรับตัวทำนายราคาหุ้น HMM คุณสามารถแสดงการสังเกตเพียงครั้งเดียวเป็นเวกเตอร์สำหรับพารามิเตอร์เหล่านี้ กล่าวคือเอ็กซ์ที= <เศษเปลี่ยน,เศษสูง,เศษต่ำ>:

นำเข้าแพนด้าเป็นคลาส pd StockPredictor(object): def __init__(self, company, n_latency_days=10): self._init_logger() self.company = company self.n_latency_days = n_latency_days self.data = pd.read_csv( 'data/company_data/ {company}.csv'.format(company=self.company)) def _init_logger(self): self._logger = logging.getLogger(__name__) handler = logging.StreamHandler() formatter = logging.Formatter( '%(asctime) s %(ชื่อ)-12s %(ชื่อระดับ)-8s %(ข้อความ)s') handler.setFormatter(ฟอร์แมต) self._logger.addHandler(ตัวจัดการ) self._logger.setLevel(logging.DEBUG) @staticmethod def _extract_features(ข้อมูล ): open_price = np.array(data['open']) close_price = np.array(data['close']) high_price = np.array(data['high']) low_price = np.array(ข้อมูล[' ต่ำ']) # คำนวณการเปลี่ยนแปลงเศษส่วนของราคาปิด สูง และต่ำ # ซึ่งจะใช้คุณลักษณะ frac_change = (close_price - open_price) / open_price frac_high = (high_price - open_price) / open_price frac_low = (open_price - low_price) / open_price กลับ np.column_stack((frac_change, frac_high, frac_low)) # ตัวทำนายสำหรับหุ้น GOOGLstock_predictor = StockPredictor(company='GOOGL')

ขั้นตอนแรกในการทำนายราคาคือการฝึกให้ HMM คำนวณพารามิเตอร์จากลำดับการสังเกตที่กำหนด เนื่องจากการสังเกตเป็นเวกเตอร์ของตัวแปรสุ่มแบบต่อเนื่อง สมมติว่าการแจกแจงความน่าจะเป็นของการปล่อยก๊าซเป็นแบบต่อเนื่อง

เพื่อความง่าย สมมติว่าเป็นการแจกแจงแบบเกาส์เซียนพหุนามพร้อมพารามิเตอร์ (มและส). ดังนั้น คุณต้องกำหนดพารามิเตอร์ต่อไปนี้สำหรับเมทริกซ์การเปลี่ยนแปลงก, ความน่าจะเป็นก่อนหน้า,ปี่, พร้อมด้วยมและสซึ่งเป็นตัวแทนของการแจกแจงแบบเกาส์แบบพหุนาม

จาก hmmlearn.hmm นำเข้า GaussianHMM class StockPredictor(object): def __init__(self, company, n_latency_days=10, n_hidden_states=4): self._init_logger() self.company = company self.n_latency_days = n_latency_days self.hmm = GaussianHMM(n_components =n_hidden_states) self.data = pd.read_csv( 'data/company_data/{company}.csv'.format(company=self.company)) def fit(self): self._logger.info('>>> คุณสมบัติการแยกข้อมูล ') feature_vector = StockPredictor._extract_features(self.data) self._logger.info('การแยกคุณสมบัติเสร็จสมบูรณ์ <<<') self.hmm.fit(feature_vector)

ในแมชชีนเลิร์นนิง คุณจะแบ่งชุดข้อมูลทั้งหมดออกเป็นสองประเภท ชุดแรกคือชุดข้อมูลการฝึกใช้เพื่อฝึกโมเดล ชุดที่สองคือชุดข้อมูลทดสอบใช้เพื่อประเมินโมเดลสุดท้ายที่เหมาะสมกับชุดข้อมูลการฝึกอบรม

การแยกชุดข้อมูลการฝึกออกจากชุดข้อมูลทดสอบจะป้องกันไม่ให้ข้อมูลมากเกินไปในโมเดล ในกรณีนี้ ให้แยกชุดข้อมูลออกเป็นสองประเภทรถไฟ_dataสำหรับฝึกนายแบบและทดสอบ_ข้อมูลสำหรับการประเมินแบบจำลอง ในการดำเนินการดังกล่าว ให้ใช้รถไฟ_test_splitวิธีการที่จัดทำโดยsklearn.model_selectionโมดูล:

จาก sklearn.model_selection นำเข้า train_test_splitclass StockPredictor(object): def __init__(self, company, test_size=0.33, n_latency_days=10, n_hidden_states=4): self._init_logger() self.company = company self.n_latency_days = n_latency_days self.hmm = GaussianHMM(n_components=n_hidden_states) self._split_train_test_data(test_size) def _split_train_test_data(self, test_size): data = pd.read_csv( 'data/company_data/{company}.csv'.format(company=self.company)) _train_data, test_data = train_test_split( data, test_size=test_size, shuffle=False) self._train_data = _train_data self._test_data = test_data def fit(self): self._logger.info('>>> Extracting Features') feature_vector = StockPredictor._extract_features(ตัวเอง ._train_data) self._logger.info('การแยกคุณสมบัติเสร็จสมบูรณ์ <<<') self.hmm.fit(feature_vector)

เดอะรถไฟ_test_splitสามารถแยกอาร์เรย์หรือเมทริกซ์ออกเป็นรถไฟสุ่มและทดสอบชุดย่อย เมื่อคุณฝึก HMM ของคุณด้วยข้อมูลตามลำดับ คุณไม่ต้องการแยกข้อมูลแบบสุ่ม เพื่อป้องกันการแยกข้อมูลการทดสอบและการฝึกอบรมแบบสุ่ม ให้ผ่านสับเปลี่ยน = เท็จเป็นข้อโต้แย้ง

เมื่อแบบจำลองของคุณได้รับการฝึกฝนแล้ว คุณต้องคาดการณ์ราคาปิดของหุ้น ดังที่ได้กล่าวไว้ก่อนหน้านี้ คุณต้องการคาดการณ์ราคาปิดของหุ้นในหนึ่งวันโดยที่คุณทราบราคาเปิด ซึ่งหมายความว่าหากคุณสามารถทำนายได้เศษเปลี่ยนในแต่ละวันสามารถคำนวณราคาปิดได้ดังนี้

ดังนั้น ปัญหาของคุณจึงจบลงที่การคำนวณเอ็กซ์เสื้อ+1= <เศษเปลี่ยน,เศษสูง,เศษต่ำ>เวกเตอร์การสังเกตสำหรับวันที่กำหนดข้อมูลการสังเกตสำหรับทีวันx1,…,xทีและพารามิเตอร์ของ HMM

ซึ่งเป็นการหาค่าของเอ็กซ์เสื้อ+1ที่เพิ่มความน่าจะเป็นหลังของพี(เอ็กซ์เสื้อ+1|X1,…,Xที,ฉัน):

เมื่อคุณลบพารามิเตอร์ทั้งหมดที่ไม่ขึ้นกับเอ็กซ์เสื้อ+1จากสมการการหาค่าสูงสุด คุณจะเหลือโจทย์ในการหาค่าของเอ็กซ์เสื้อ+1ซึ่งปรับความน่าจะเป็นให้เหมาะสมที่สุดพี(เอ็กซ์1,…,Xเสื้อ+1ฉัน). ถ้าสมมุติว่าเศษเปลี่ยนเป็นตัวแปรต่อเนื่อง การเพิ่มประสิทธิภาพของปัญหาจะทำได้ยากในการคำนวณ

ดังนั้น ให้แบ่งการเปลี่ยนแปลงที่เป็นเศษส่วนเหล่านี้ออกเป็นค่าที่ไม่ต่อเนื่องระหว่างตัวแปรจำกัดสองตัว (ดังที่ระบุไว้ในตารางต่อไปนี้) และค้นหาชุดของการเปลี่ยนแปลงที่เป็นเศษส่วน<เศษเปลี่ยน,เศษสูง,เศษต่ำ>ที่จะเพิ่มความน่าจะเป็นพี(เอ็กซ์1,…,Xเสื้อ+1ฉัน):

ดังนั้น ด้วยชุดค่าที่ไม่ต่อเนื่องก่อนหน้านี้ ให้รัน (20 x 10 x 10 =) 2,000 การดำเนินการ:

def _compute_all_possible_outcomes (ตัวเอง, n_steps_frac_change, n_steps_frac_high, n_steps_frac_low): frac_change_range = np.linspace(-0.1, 0.1, n_steps_frac_change) frac_high_range = np.linspace(0, 0.1, n_steps_frac_high) frac _low_range = np.linspace(0, 0.1, n_steps_frac_low) ตัวเอง ._possible_outcomes = np.array (รายการ (itertools.product ( frac_change_range, frac_high_range, frac_low_range)))

def _get_most_probable_outcome(ตัวเอง, day_index): Previous_data_start_index = สูงสุด(0, day_index - self.n_latency_days) Previous_data_end_index = สูงสุด(0, day_index - 1) Previous_data = self._test_data.iloc[previous_data_end_index: Previous_data_end_index] Previous_data_features = StockPredictor._extract _features(ก่อนหน้า_data) Outcome_score = [] สำหรับ possible_outcome ใน self._possible_outcomes: total_data = np.row_stack( ((previous_data_features, possible_outcome)) Outcome_score.append(self.hmm.score(total_data)) most_probable_outcome = self._possible_outcomes[np.argmax(outcome_score)] ย้อนกลับ most_probable_outcome def คาดการณ์_close_price(ตัวเอง, day_index): open_price = self._test_data.iloc[day_index]['open'] Predicted_frac_change, _, _ = self._get_most_probable_outcome(day_index) return open_price * (1 + Predicted_frac_change)

"""การใช้งาน: analyse_data.py --company="""นำเข้าคำเตือน simport loggingimport itertoolsimport pandas as pdimport numpy as npimport matplotlib.pyplot as pltfrom hmmlearn.hmm import GaussianHMMfrom sklearn.model_selection import train_test_splitfrom tqdm import tqdmfrom docopt import docopt หาเรื่อง = docopt(doc=__doc__, argv=None, help=True, version=None, options_first=False) # ระงับการเตือนใน hmmlearnwarnings.filterwarnings("ignore")# เปลี่ยนรูปแบบพล็อตเป็น ggplot (เพื่อการแสดงภาพที่สวยงามและสวยงามยิ่งขึ้น)plt .style.use('ggplot') class StockPredictor(object): def __init__(ตัวเอง, บริษัท, test_size=0.33, n_hidden_states=4, n_latency_days=10, n_steps_frac_change=50, n_steps_frac_high=10, n_steps_frac_low=10): self._init_logger () self.company = บริษัท self.n_latency_days = n_latency_days self.hmm = GaussianHMM(n_components=n_hidden_states) self._split_train_test_data(test_size) self._compute_all_possible_outcomes( n_steps_frac_change, n_steps_frac_high, n_steps_frac_low) def _init_logger(ตัวเอง): self._logger = การบันทึก getLogger(__name__) handler = logging.StreamHandler() formatter = logging.Formatter( '%(asctime)s %(name)-12s %(levelname)-8s %(message)s') handler.setFormatter(formatter) self. _logger.addHandler(ตัวจัดการ) self._logger.setLevel(logging.DEBUG) def _split_train_test_data(ตัวเอง, test_size): data = pd.read_csv( 'data/company_data/{company}.csv'.format(company=self.company) ) _train_data, test_data = train_test_split( ข้อมูล, test_size=test_size, shuffle=False) self._train_data = _train_data self._test_data = test_data @staticmethod def _extract_features(ข้อมูล): open_price = np.array(data['open']) close_price = np.array(data['close']) high_price = np.array(data['high']) low_price = np.array(data['low']) # คำนวณการเปลี่ยนแปลงเศษส่วนในราคาปิด สูง และต่ำ # ซึ่งจะใช้คุณลักษณะ frac_change = (close_price - open_price) / open_price frac_high = (high_price - open_price) / open_price frac_low = (open_price - low_price) / open_price return np.column_stack((frac_change, frac_high, frac_low)) def fit(self ): self._logger.info('>>> การแยกคุณลักษณะ') feature_vector = StockPredictor._extract_features(self._train_data) self._logger.info('การแยกคุณลักษณะเสร็จสมบูรณ์ <<<') self.hmm.fit(feature_vector) def _compute_all_possible_outcomes (ตัวเอง, n_steps_frac_change, n_steps_frac_high, n_steps_frac_low): frac_change_range = np.linspace(-0.1, 0.1, n_steps_frac_change) frac_high_range = np.linspace(0, 0.1, n_steps_frac_high) frac_low _range = np.linspace(0, 0.1, n_steps_frac_low) ตัวเอง _possible_outcomes = np.array(รายการ(itertools.product( frac_change_range, frac_high_range, frac_low_range))) def _get_most_probable_outcome(self, day_index): Previous_data_start_index = max(0, day_index - self.n_latency_days) Previous_data_end_index = สูงสุด(0, day_index - 1) Previous_data = self._test_data.iloc[previous_data_end_index: Previous_data_start_index] Previous_data_features = StockPredictor._extract_features( Previous_data) Outcome_Score = [] สำหรับผลลัพธ์ที่เป็นไปได้ในตนเอง_Possible_Outcomes: Total_data = np.row_stack ( (Previous_data_features, possible_outcome)) Outcome_score.append (self.hmm .score(total_data)) most_probable_outcome = self._possible_outcomes[np.argmax(outcome_score)] ส่งคืน most_probable_outcome def ทำนาย_close_price(ตัวเอง, day_index): open_price = self._test_data.iloc[day_index]['เปิด'] ทำนาย_frac_change, _, _ = self._get_most_probable_outcome( day_index) ส่งคืน open_price * (1 + Predicted_frac_change) def defpredict_close_prices_for_days(self, days, with_plot=False): Predicted_close_prices = [] for day_index in tqdm(range(days)): Predicted_close_prices.append(self.predict_close_price(day_index) )) ถ้า with_plot: test_data = self._test_data[0: วัน] วัน = np.array(test_data['date'], dtype="datetime64[ms]") actual_close_prices = test_data['close'] fig = plt.figure แกน set_title('{company}'.format(company=self.company)) fig.autofmt_xdate() plt.legend() plt.show() return modified_close_prices stock_predictor = StockPredictor(company=args['--company'])stock_predictor .fit()stock_predictor.predict_close_prices_for_days(500, with_plot=True)

ผลลัพธ์จะเป็นดังนี้:

บทสรุป

คุณคาดการณ์ราคาหุ้นสำเร็จโดยใช้ HMM คุณใช้วิธีการประมาณค่าพารามิเตอร์และการประเมินแบบจำลองเพื่อกำหนดราคาปิดของหุ้น การใช้ HMM ในการวิเคราะห์ตลาดหุ้นเป็นเพียงอีกตัวอย่างหนึ่งของการประยุกต์ใช้ HMM ในการวิเคราะห์ข้อมูลอนุกรมเวลา

ขอบคุณสำหรับการอ่าน!