## 回測環境開發流程
#### <font style="color: #30598c;">🐞完整程式碼範例</font>
<a class="button" href="https://drive.google.com/file/d/1UonyRRj5Q1zMkxv8wbVrLofDWI7r_8de/view?usp=sharing">
<font class="button_text">程式碼下載</font>
</a>
</br>
#### 1. 下載歷史股價
```python
def fetch_history_data(stock_code: str, start_date: str, end_date: str) -> pd.DataFrame:
"""
獲取歷史股價資料
"""
# 透過 yfinance 下載歷史股價
source_data = yf.download(tickers=f"{stock_code}.TW", start=start_date, end=end_date, auto_adjust=False)
# 回傳下載後的資料
return source_data
```
</br>
#### 2. 調整資料欄位 (def format_data_columns)
```python
def format_data_columns(data: pd.DataFrame) -> pd.DataFrame:
"""
格式化資料欄位
"""
# 取出欄位名稱的最上層
data.columns = data.columns.get_level_values(0)
# 去掉欄位名稱 (Price)
data.columns.name = None
# 重設索引，將 Date 從 index 轉成欄位
pre_data = data.reset_index()
# 依需求的欄位順序
pre_data = pre_data[["Date", "Open", "High", "Low", "Close", "Volume"]]
# 回傳格式化後的資料
return pre_data
```
#### <font style="color: #30598c;">🐞如果沒有辦法成功透過 `yfinance` 獲取資料，點擊連結下載</font>
<a class="button" href="https://drive.google.com/file/d/1zX0sczmbLkC35UcbQ72Loodo-P_7guhJ/view?usp=sharing">
<font class="button_text">歷史資料下載</font>
</a>
</br>
#### 3. 計算技術指標 (def fetch_TA_data)
```python
def fetch_TA_data(data: pd.DataFrame) -> pd.DataFrame:
"""
計算技術指標
"""
# 計算移動平均線
data["MA10"] = data["Close"].rolling(window=10).mean()
data["MA60"] = data["Close"].rolling(window=60).mean()
# 四捨五入到小數點後兩位
ta_df = data.round(2)
# 回傳包含技術指標的資料
return ta_df
```
</br>
#### 4. 資料逐筆回傳 (def recieve_kBar)
```python
# 將資料逐筆拆分給回測系統
for index, row in ta_data.iterrows():
# 將 Series 轉成 DataFrame 格式
chart = row.to_frame().T.reset_index(drop=True)
# 將資料傳遞給回測系統
run_backtest(chart)
```
</br>
#### 5. 新增接收行情通道
```python
def receive_kBar(chart: pd.DataFrame):
"""
執行回測
"""
# 打印行情資訊
cs.print("接收行情資訊:", style="bold yellow")
cs.print(f"{chart}\n")
```
</br>
#### 6. 判斷交易信號 (fetch_trade_singnal)
```python
# 交易信號: Buy, Sell, Clear, Hold
trade_signal = "Hold"
```
```python
def fetch_trade_signal(chart: pd.DataFrame):
"""
獲取交易信號
"""
global trade_signal, trade_signal_data, trade_order
# 將最新 k 棒資料加入交易信號資料集
trade_signal_data = pd.concat([trade_signal_data, chart], ignore_index=True)
# 確保資料數大於 2 以便後續判斷
if len(trade_signal_data) < 2:
trade_signal = "Hold"
return
# 僅保留最近兩筆資料
trade_signal_data = trade_signal_data.tail(2)
# 擷取今日、昨日 k 棒資訊
curr_date = trade_signal_data.iloc[-1]["Date"]
prev_close = trade_signal_data.iloc[0]["Close"]
curr_close = trade_signal_data.iloc[-1]["Close"]
prev_ma10 = trade_signal_data.iloc[0]["MA10"]
curr_ma10 = trade_signal_data.iloc[-1]["MA10"]
curr_ma60 = trade_signal_data.iloc[-1]["MA60"]
# 更新委託單資訊
trade_order["date"] = curr_date.strftime("%Y-%m-%d")
trade_order["price"] = float(curr_close)
# 判斷買賣訊號
if curr_close < curr_ma60:
trade_signal = "Clear"
elif (prev_close <= prev_ma10) and (curr_close > curr_ma10) and (position_count == 0):
trade_signal = "Buy"
elif (prev_close >= prev_ma10) and (curr_close < curr_ma10) and (position_count > 0):
trade_signal = "Sell"
else:
trade_signal = "Hold"
```
</br>
#### 7. 執行交易 (trading)
```python
# 建立委託單格式
trade_order = {"action": None, "price": 0, "quantity": 0, "fee": 0, "tax": 0, "ar": 0, "date": None}
```
```python
def trading() -> str:
"""
執行交易
"""
global position_count
# 根據交易信號決定交易動作
if trade_signal == "Buy":
action = "Buy"
trade_order["quantity"] = 1
elif trade_signal == "Sell":
action = "Sell"
trade_order["quantity"] = 1
elif trade_signal == "Clear" and position_count > 0:
action = "Sell"
trade_order["quantity"] = position_count
else:
action = "Hold"
return action
# 更新委託單資訊
trade_order["action"] = action
# 計算委託價金
cost = trade_order["price"] * trade_order["quantity"] * 1000
# 計算交易稅
if action == "Buy":
trade_order["tax"] = 0
else:
trade_order["tax"] = int(cost * 0.003)
# 計算手續費
trade_order["fee"] = int(cost * 0.001425)
trade_order["ar"] = cost + trade_order["fee"] + trade_order["tax"]
# 紀錄整張委託單
trade_order_list.append(trade_order.copy())
# 更新庫存數量
if trade_signal == "Buy":
position_count += 1
elif trade_signal == "Sell":
position_count -= 1
else:
position_count = 0
# 回傳交易動作
return action
```
</br>
#### 8. 更新損益紀錄
```python
# 建立損益紀錄格式
trade_record = {"trade_count": {"Buy": 0, "Sell": 0, "Total": 0}, "pnl": {"Realized": 0, "Unrealized": 0}, "pnl_count": {"Profit": 0, "Loss": 0, "Total": 0}, "transaction_cost": {"Fee": 0, "Tax": 0, "Total": 0}}
```
```python
def update_pnl_record(action: str):
"""
更新交易、損益資訊
"""
global backtest_capital, trade_record
# 更新回測本金
if action == "Buy":
backtest_capital -= trade_order["ar"]
elif action == "Sell":
backtest_capital += trade_order["ar"]
else:
return
# 更新交易次數
trade_record["trade_count"][action] += 1
trade_record["trade_count"]["Total"] = trade_record["trade_count"]["Buy"] + trade_record["trade_count"]["Sell"]
# 更新已實現損益紀錄
if action == "Buy":
trade_record["pnl"]["Unrealized"] += trade_order["ar"]
# 更新未實現損益紀錄
else:
curr_realized = (trade_order["ar"] - trade_record["pnl"]["Unrealized"])
trade_record["pnl"]["Realized"] += curr_realized
trade_record["pnl"]["Unrealized"] = 0
# 更新獲利/虧損次數
if curr_realized > 0:
trade_record["pnl_count"]["Profit"] += 1
elif curr_realized < 0:
trade_record["pnl_count"]["Loss"] += 1
trade_record["pnl_count"]["Total"] = trade_record["pnl_count"]["Profit"] + trade_record["pnl_count"]["Loss"]
# 更新交易成本
trade_record["transaction_cost"]["Fee"] += trade_order["fee"]
trade_record["transaction_cost"]["Tax"] += trade_order["tax"]
trade_record["transaction_cost"]["Total"] = trade_record["transaction_cost"]["Fee"] + trade_record["transaction_cost"]["Tax"]
```
<style>
.tips {
display: flex;
gap: 2px;
padding-bottom: 16px;
}
.tips-content {
font-family: "Noto Sans TC";
color: #3f628a;
font-weight: 600;
}
.button {
display: flex;
justify-content: center;
background-color: #3f628a;
padding: 3px 5px 5px 5px;
width: 20%;
border-radius: 6px;
}
.red_btn {
background-color: #b00d0d;
}
.button_text {
font-family: "Noto Sans TC";
font-weight: 500;
color: #FFFFFF;
}
</style>
Sign in with Wallet
Connect another wallet