建立馬可夫決策交易環境以模擬股票市場行為

利用Python實作馬可夫決策過程，最大化股市投資利益並顯示買賣信號

三大重點概要

• 環境設定與參數定義：設置交易環境的基本參數，包括手續費、稅率、買入單位及初始資金。
• 馬可夫決策過程模型建立：定義狀態、動作及獎勵函數，並使用值迭代尋求最佳策略。
• 買賣信號生成與K線圖展示：根據模型策略生成買賣信號，並將其標註於K線圖上以視覺化交易決策。

1. 環境設定與參數定義

1.1. 參數設置

首先，定義交易環境的基本參數，包括最低手續費（MIN_FEE）、手續費率（FEE_RATE）、賣出稅率（SELL_TAX_RATE）、每次買入的最小股數（BUY_UNIT）、初始資金（START_CASH），以及研究的股票代號（STOCK_CODE）。這些參數將構成交易環境的基礎設定。

MIN_FEE = 20  # 最低手續費
FEE_RATE = 0.001425  # 手續費率
SELL_TAX_RATE = 0.003  # 賣出稅率
BUY_UNIT = 1000  # 每次買入單位股數
START_CASH = 1000000  # 初始資金
STOCK_CODE = '2609.TW'  # 股票代號

1.2. 載入必要的Python套件

使用Python進行馬可夫決策過程的實作，需要導入多個套件來處理數據、繪圖以及建立交易環境。

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import yfinance as yf
from datetime import datetime, timedelta
import gym
from gym import spaces
import mplfinance as mpf

2. 下載與處理股票數據

2.1. 下載股票歷史數據

使用yfinance套件下載指定股票的歷史價格數據，並計算每日的報酬率（Returns）。以下範例以2024年1月1日至2025年1月23日的數據為例。

end_date = datetime(2025, 1, 23)  # 假設今天是2025年1月23日
start_date = datetime(2024, 1, 1)  # 從2024年1月1日開始
stock_data = yf.download(STOCK_CODE, start=start_date, end=end_date)
stock_data['Returns'] = stock_data['Adj Close'].pct_change()  # 計算每日報酬率
stock_data.dropna(inplace=True)  # 移除NaN值

2.2. 可視化股票價格與報酬率

繪製股票的收盤價與報酬率，幫助理解數據的變動趨勢。

plt.figure(figsize=(14, 7))
plt.subplot(2, 1, 1)
plt.plot(stock_data['Adj Close'], label='Adj Close Price', color='blue')
plt.title(f'{STOCK_CODE} 調整後收盤價')
plt.xlabel('日期')
plt.ylabel('價格')
plt.legend()

plt.subplot(2, 1, 2)
plt.plot(stock_data['Returns'], label='Daily Returns', color='orange')
plt.title('每日報酬率')
plt.xlabel('日期')
plt.ylabel('報酬率')
plt.legend()

plt.tight_layout()
plt.show()

3. 定義馬可夫決策過程（MDP）

3.1. 創建自定義交易環境類

使用Gym庫創建一個自定義的交易環境類StockTradingEnv，繼承自gym.Env，並定義其狀態空間、動作空間及獎勵函數。

class StockTradingEnv(gym.Env):
    def __init__(self, df, start_cash=START_CASH, min_fee=MIN_FEE, fee_rate=FEE_RATE, sell_tax=SELL_TAX_RATE, buy_unit=BUY_UNIT):
        super(StockTradingEnv, self).__init__()
        self.df = df
        self.start_cash = start_cash
        self.min_fee = min_fee
        self.fee_rate = fee_rate
        self.sell_tax = sell_tax
        self.buy_unit = buy_unit
        self.current_step = 0
        self.cash = self.start_cash
        self.holdings = 0
        self.total_value = self.cash

        # 定義動作空間：0=持有, 1=買入, 2=賣出
        self.action_space = spaces.Discrete(3)

        # 定義觀察空間：[當前價格，現金餘額，持股數]
        self.observation_space = spaces.Box(low=0, high=np.inf, shape=(3,), dtype=np.float32)

    def reset(self):
        self.current_step = 0
        self.cash = self.start_cash
        self.holdings = 0
        self.total_value = self.cash
        return self._get_state()

    def _get_state(self):
        current_price = self.df.iloc[self.current_step]['Close']
        return np.array([current_price, self.cash, self.holdings])

    def calculate_reward(self, action, current_price, next_price):
        reward = 0
        if action == 1:  # 買入
            cost = current_price * self.buy_unit + max(self.min_fee, current_price * self.buy_unit * self.fee_rate)
            if self.cash >= cost:
                self.cash -= cost
                self.holdings += self.buy_unit
            else:
                reward = -100  # 現金不足懲罰
        elif action == 2:  # 賣出
            if self.holdings >= self.buy_unit:
                revenue = current_price * self.buy_unit * (1 - self.sell_tax) - max(self.min_fee, current_price * self.buy_unit * self.fee_rate)
                self.cash += revenue
                self.holdings -= self.buy_unit
                reward = revenue  # 賣出收益
            else:
                reward = -100  # 持股不足懲罰
        return reward

    def step(self, action):
        current_price = self.df.iloc[self.current_step]['Close']
        reward = self.calculate_reward(action, current_price, current_price)  # 簡化為當前價格
        self.current_step += 1
        done = self.current_step >= len(self.df) - 1
        next_state = self._get_state()
        self.total_value = self.cash + self.holdings * current_price
        return next_state, reward, done, {}

3.2. 定義狀態轉移矩陣與獎勵函數

假設狀態為股票的漲跌（1: 上漲，0: 下跌），動作為買入、賣出或持有，並定義狀態轉移矩陣及獎勵函數以反映交易行為的收益。

# 定義狀態轉移矩陣：上漲後繼續上漲的機率為0.6，下跌的機率為0.4；下跌後繼續下跌的機率為0.7，上漲的機率為0.3
    transition_matrix = np.array([
        [0.6, 0.4],
        [0.3, 0.7]
    ])

    # 定義獎勵矩陣
    rewards = [
        [10, -10],  # 買入的獎勵
        [-10, 10],  # 賣出的獎勵
        [0, 0]       # 持有的獎勵
    ]

3.3. 值迭代算法尋求最佳策略

使用值迭代算法來計算每個狀態的最佳價值，從而導出最優策略。

def value_iteration(transition_matrix, rewards, gamma=0.9, epsilon=1e-6):
    num_states = transition_matrix.shape[0]
    num_actions = len(rewards)
    V = np.zeros(num_states)
    while True:
        delta = 0
        for s in range(num_states):
            Q = [sum([transition_matrix[s][s1] * (rewards[a][s] + gamma * V[s1]) for s1 in range(num_states)]) for a in range(num_actions)]
            max_q = max(Q)
            delta = max(delta, abs(max_q - V[s]))
            V[s] = max_q
        if delta < epsilon:
            break
    return V

# 執行值迭代
optimal_values = value_iteration(transition_matrix, rewards)

4. 實作交易策略與模擬交易

4.1. 初始化交易狀態

設置初始資金與持股數，並準備一個列表來存儲交易信號。

cash = START_CASH
shares = 0
signals = []

4.2. 模擬交易過程

根據最佳策略模擬交易，並根據狀態轉移與獎勵函數生成買賣信號。

for i in range(1, len(stock_data)):
    current_state = 1 if stock_data['Returns'].iloc[i] > 0 else 0
    Q = [sum([transition_matrix[current_state][s1] * (rewards[a][current_state] + 0.9 * optimal_values[s1]) for s1 in range(2)]) for a in range(3)]
    action = np.argmax(Q)
    action_label = ["buy", "sell", "hold"][action]
    signals.append(action_label)

4.3. 將交易信號加入數據框

將生成的交易信號加入到股票數據框中，便於後續的可視化處理。

stock_data['Signal'] = [None] + signals

5. 顯示買賣信號於K線圖上

5.1. 繪製K線圖與買賣信號

使用mplfinance庫繪製股票的K線圖，並在圖上標註買入與賣出信號。

plt.figure(figsize=(14, 7))
plt.plot(stock_data['Adj Close'], label='價格', color='blue')

# 標註買入信號
buy_signals = stock_data[stock_data['Signal'] == 'buy']
plt.scatter(buy_signals.index, buy_signals['Adj Close'], label='買入', color='green', marker='^', alpha=1)

# 標註賣出信號
sell_signals = stock_data[stock_data['Signal'] == 'sell']
plt.scatter(sell_signals.index, sell_signals['Adj Close'], label='賣出', color='red', marker='v', alpha=1)

plt.title(f'{STOCK_CODE} 股票買賣信號')
plt.xlabel('日期')
plt.ylabel('價格')
plt.legend()
plt.show()

5.2. 使用mplfinance繪製更詳細的K線圖

整合K線圖與買賣信號，提供更直觀的視覺化效果。

apds = [
    mpf.make_addplot(buy_signals['Adj Close'], type='scatter', markersize=100, marker='^', color='green'),
    mpf.make_addplot(sell_signals['Adj Close'], type='scatter', markersize=100, marker='v', color='red')
]

mpf.plot(stock_data, type='candle', addplot=apds, volume=True, style='charles', title=f'{STOCK_CODE} K線圖與買賣信號')

6. 結果分析與優化建議

6.1. 結果分析

觀察K線圖上的買賣信號，可以評估交易策略的有效性。綠色三角形標記買入點，紅色倒三角形標記賣出點。透過這些信號，可以分析策略在不同市場情況下的表現。

6.2. 策略優化建議

目前的模型基於簡化的狀態轉移矩陣與獎勵函數，未考慮更多市場因素。建議進一步優化策略，例如：

• 引入更多技術指標作為狀態變量，如移動平均線、相對強弱指數（RSI）等。
• 採用更複雜的獎勵函數，考慮風險調整後的收益。
• 結合強化學習算法，如Q-Learning或深度強化學習，以提升策略的適應性與預測能力。

7. 參考資料

1. Markov Decision Processes and Reinforcement Learning in Stock Trading
2. Markov decision process: value iteration with code implementation
3. Markov Decision Process - GeeksforGeeks
4. Markov Chains in Stock Market Using Python - Getting Transition Matrix
5. Automated stock trading using FinRL with financial data
6. Markov Decision Process Toolbox for Python
7. Reinforcement Learning Stock Trading
8. Deep Reinforcement Learning in Stock Trading
9. 量化投資與馬可夫決策過程應用
10. 基於馬可夫決策過程的股票交易策略實現