其實一直以來都很喜歡研究量化交易 自動化交易 或是任何用程式邏輯來做交易的領域
但是一直沒做出一個很大型的python套件 或是持久運行的網站最近給自己一個目標 來寫部落格 從零開始記錄自己實現每個元件的過程
我的想法是先從每個元件功能開始著手 例如做一個"策略"類別
策略類別是一個父類別 又可以衍伸出 MACD策略 RSI策略等等
然後實例化一個策略之後可以用這個實例化之後的策略來產生交易訊號
甚至把策略丟到一個回測函數 可以產生回測結果
是的我打算每個元件自己慢慢刻
最後整合的事情才交給AI
包括整合成一包python套件 開源的任何人都可以pip install安裝
甚至整合前端跟資料庫做成一個網站服務
最終目標希望是做出真的有用的程式能夠提供準確判斷賺錢
萬事起頭難 先整理一下自己會的技術
我目前會的有基於規則的交易
例如玩股網的選股百寶袋功能我幾乎都能python山寨出類似的東西
基於監督學習的交易
例如給CNN神經網路看一張"圖"讓他告訴我該買還是該賣
基於強化學習的交易
定義好模擬的交易環境 獎勵函數 強化學習算法
丟台GPU給他訓練
自動化交易
利用github workflow或render webservice 自動化爬資料 預測買賣 自動下單
目前感覺許多技術都有碰過
但一直沒有好好整合所以這篇是當起頭 開始做這個專案
至於專案名稱好多很酷的名字都有人取過了 邊做邊想吧
第一章手搓策略
什麼是策略 想像一下 常見的技術指標 MACD RSI等等
如果用這些技術指標判斷進出場的策略 在python理應該怎麼定義跟使用
首先創建一個通用策略父類別 (BaseStrategy)
from abc import ABC, abstractmethod
import pandas as pd
class BaseStrategy(ABC):
def __init__(self, data: pd.DataFrame):
"""
:param data: 包含收盤價等欄位的歷史資料 DataFrame
"""
self.data = data
self.signals = pd.Series(index=self.data.index, dtype="object")
@abstractmethod
def generate_signals(self):
"""
子類別實作策略邏輯，填入 self.signals（'buy', 'sell', 'hold'）
"""
pass
def get_signals(self) -> pd.Series:
return self.signals
MACD 策略子類別 (MACDStrategy)
class MACDStrategy(BaseStrategy):
def __init__(self, data: pd.DataFrame, fast=12, slow=26, signal=9):
super().__init__(data)
self.fast = fast
self.slow = slow
self.signal_period = signal
def generate_signals(self):
close = self.data['close']
ema_fast = close.ewm(span=self.fast, adjust=False).mean()
ema_slow = close.ewm(span=self.slow, adjust=False).mean()
macd = ema_fast - ema_slow
signal_line = macd.ewm(span=self.signal_period, adjust=False).mean()
self.signals = pd.Series(index=self.data.index, dtype="object")
self.signals[macd > signal_line] = 'buy'
self.signals[macd < signal_line] = 'sell'
self.signals[macd == signal_line] = 'hold'
RSI 策略子類別 (RSIStrategy)
class RSIStrategy(BaseStrategy):
def __init__(self, data: pd.DataFrame, period=14, overbought=70, oversold=30):
super().__init__(data)
self.period = period
self.overbought = overbought
self.oversold = oversold
def generate_signals(self):
delta = self.data['close'].diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
avg_gain = gain.rolling(self.period).mean()
avg_loss = loss.rolling(self.period).mean()
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
self.signals = pd.Series(index=self.data.index, dtype="object")
self.signals[rsi < self.oversold] = 'buy'
self.signals[rsi > self.overbought] = 'sell'
self.signals[(rsi >= self.oversold) & (rsi <= self.overbought)] = 'hold'
使用方式
import yfinance as yf
import datetime
import pandas as pd
# 下載多幣種也沒問題
df = yf.download("BTC-USD", start="2023-01-01", end=datetime.datetime.now())
# 通用欄位標準化
if isinstance(df.columns, pd.MultiIndex):
df.columns = ['_'.join([str(c).lower() for c in col]) for col in df.columns]
tickers = set(col.split('_')[-1] for col in df.columns)
if len(tickers) == 1:
ticker_suffix = list(tickers)[0]
df.columns = [col.replace(f"_{ticker_suffix}", '') for col in df.columns]
else:
df.columns = df.columns.str.lower()
# 檢查是否有 close
assert 'close' in df.columns, "找不到 close 欄位"
# 套用策略
strategy = MACDStrategy(df)
strategy.generate_signals()
strategy.get_signals()
執行結果
import matplotlib.pyplot as plt
signals = strategy.get_signals()
close = df['close']
plt.figure(figsize=(14, 7))
plt.plot(close.index, close, label='Close Price', color='black')
# 買點用綠色三角形標記
buy_signals = signals == 'buy'
plt.scatter(close.index[buy_signals], close[buy_signals], marker='^', color='green', label='Buy Signal', s=100)
# 賣點用紅色倒三角形標記
sell_signals = signals == 'sell'
plt.scatter(close.index[sell_signals], close[sell_signals], marker='v', color='red', label='Sell Signal', s=100)
plt.title('Price Chart with Buy/Sell Signals')
plt.xlabel('Date')
plt.ylabel('Price')
plt.legend()
plt.grid(True)
plt.show()
畫圖
完整程式碼
https://github.com/skywalker0803r/AlphaCoin/tree/main
