Python 程式交易入門:50 行程式碼打造自動下單機器人|2026 完整教學
快速導覽: 50行程式碼 | API連接教學 | 安全警示 | 風險控制 | 常見問題
Hook:50 行?真的假的?
聽過太多「量化交易很難」的說法?
事實是,Python 程式交易的入門門檻,比你想像中低得多。不需要金融工程碩士學位,不需要花十萬塊上課,只需要 50 行簡潔的程式碼,你就能打造一個真正會自動下單的交易機器人。
本文將帶你從零開始,一步步實作一個能連接交易所 API、自動判斷訊號、執行買賣的完整系統。準備好了嗎?
程式交易基礎概念
什麼是程式交易?
程式交易(Algorithmic Trading)指的是用程式碼自動執行交易策略,取代人工看盤、手動下單。它的核心優勢包括:
- 速度:毫秒級反應,遠超人類極限
- 紀律:完全依照策略執行,不受情緒干擾
- 回測:可用歷史數據驗證策略有效性
- 多工:同時監控數十個標的,24/7 不間斷
程式交易的基本流程
取得市場數據 → 計算技術指標 → 產生交易訊號 → 執行下單 → 記錄與監控
為什麼選擇 Python?
| 優勢 | 說明 |
|------|------|
| 語法簡潔 | 接近自然語言,新手友善 |
| 生態豐富 | pandas、numpy、ta-lib 等金融分析套件 |
| API 支援 | 各大交易所都提供 Python SDK |
| 社群活躍 | 問題易找到解答,資源充足 |
50 行完整程式碼(含詳細註解)
以下是一個完整可運作的趨勢跟蹤機器人,使用雙均線交叉策略:
import ccxt
import pandas as pd
import time
from datetime import datetime
# ========== 設定參數 ==========
# ⚠️ 安全警告:絕對不要將 API 金鑰直接寫在程式碼中!
# 請使用環境變數或安全的金鑰管理服務
API_KEY = 'your_api_key_here' # 從 Binance API 管理頁面取得
API_SECRET = 'your_api_secret_here' # Secret Key 只顯示一次,請妥善保存
SYMBOL = 'BTC/USDT' # 交易對:比特幣兌 USDT
TIMEFRAME = '1h' # K線週期:1小時
FAST_MA = 10 # 快線週期:10 根 K 線
SLOW_MA = 30 # 慢線週期:30 根 K 線
POSITION_SIZE = 0.001 # 每次下單數量:0.001 BTC
# ========== 初始化交易所 ==========
# 使用 CCXT 函式庫連接 Binance
# enableRateLimit: 自動處理 API 頻率限制,避免被封鎖
exchange = ccxt.binance({
'apiKey': API_KEY,
'secret': API_SECRET,
'enableRateLimit': True,
'options': {'defaultType': 'spot'} # 現貨交易模式
})
# ========== 取得 K 線數據 ==========
def get_ohlcv(symbol, timeframe, limit=100):
"""
取得歷史 K 線數據(Open, High, Low, Close, Volume)
參數:
symbol: 交易對,如 'BTC/USDT'
timeframe: 時間週期,如 '1h', '4h', '1d'
limit: 取得多少根 K 線,預設 100 根
回傳:
DataFrame 包含 timestamp, open, high, low, close, volume
"""
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
return df
# ========== 計算技術指標 ==========
def calculate_signals(df):
"""
計算雙均線交叉訊號
邏輯說明:
- 快線 > 慢線:看多訊號(signal = 1)
- 快線 < 慢線:看空訊號(signal = -1)
- 快線 = 慢線:無訊號(signal = 0)
這是最基礎的趨勢跟蹤策略,適合初學者理解
"""
# 計算簡單移動平均線(SMA)
df['fast_ma'] = df['close'].rolling(window=FAST_MA).mean()
df['slow_ma'] = df['close'].rolling(window=SLOW_MA).mean()
# 初始化訊號欄位
df['signal'] = 0
df.loc[df['fast_ma'] > df['slow_ma'], 'signal'] = 1 # 黃金交叉 → 買入
df.loc[df['fast_ma'] < df['slow_ma'], 'signal'] = -1 # 死亡交叉 → 賣出
return df
# ========== 執行下單 ==========
def execute_trade(signal, symbol, amount):
"""
根據訊號執行買賣操作
參數:
signal: 1 為買入, -1 為賣出
symbol: 交易對
amount: 交易數量
回傳:
order 物件(成功)或 None(失敗)
"""
try:
if signal == 1:
# 市價買入:以當前市場最優價格立即成交
order = exchange.create_market_buy_order(symbol, amount)
print(f"[{datetime.now()}] 🟢 買入 {amount} {symbol}")
return order
elif signal == -1:
# 市價賣出:以當前市場最優價格立即成交
order = exchange.create_market_sell_order(symbol, amount)
print(f"[{datetime.now()}] 🔴 賣出 {amount} {symbol}")
return order
except Exception as e:
print(f"[{datetime.now()}] ❌ 下單失敗: {e}")
return None
# ========== 主程式 ==========
def main():
"""
主循環:每小時檢查一次訊號
執行流程:
1. 取得最新 K 線數據
2. 計算均線與訊號
3. 訊號改變時執行交易
4. 等待下一個週期
"""
print(f"🚀 交易機器人啟動 | 交易對: {SYMBOL} | 時間框架: {TIMEFRAME}")
last_signal = 0
while True:
try:
# 取得數據並計算訊號
df = get_ohlcv(SYMBOL, TIMEFRAME)
df = calculate_signals(df)
current_signal = df['signal'].iloc[-1]
# 訊號改變時才執行交易(避免重複下單)
if current_signal != last_signal and current_signal != 0:
execute_trade(current_signal, SYMBOL, POSITION_SIZE)
last_signal = current_signal
# 輸出目前狀態
print(f"[{datetime.now()}] 快線: {df['fast_ma'].iloc[-1]:.2f} | 慢線: {df['slow_ma'].iloc[-1]:.2f} | 訊號: {current_signal}")
# 等待下一個週期(3600 秒 = 1 小時)
time.sleep(3600)
except Exception as e:
print(f"[{datetime.now()}] ⚠️ 錯誤: {e}")
time.sleep(60) # 錯誤後等待 1 分鐘重試
# 啟動機器人
if __name__ == '__main__':
main()
程式碼區塊解析
| 區塊 | 功能說明 | 關鍵技術點 |
|------|----------|------------|
| 設定參數 | 集中管理所有可調整的變數 | 使用大寫常數命名慣例 |
| 初始化交易所 | 使用 ccxt 連接 Binance | enableRateLimit 防止 API 被封 |
| 取得 K 線 | 抓取歷史價格數據 | fetch_ohlcv() 方法 |
| 計算訊號 | 雙均線交叉產生買賣訊號 | Pandas rolling() 與 mean() |
| 執行下單 | 根據訊號自動買賣 | create_market_buy_order() |
| 主程式 | 無限循環,定時檢查 | while True + time.sleep() |
如何連接交易所 API
Step 1:申請 API 金鑰(以 Binance 為例)
- 登入 Binance 帳號
- 進入「API 管理」頁面
- 創建新的 API Key
- 務必啟用 IP 白名單,限制只有你的伺服器能存取
- 記下 API Key 和 Secret(Secret 只顯示一次)
Step 2:安裝必要套件
# 安裝 CCXT(支援 100+ 交易所的統一 API)
# 安裝 Pandas(數據處理)
pip install ccxt pandas python-dotenv
Step 3:測試連線
import ccxt
# 測試連線(不帶餘額查詢,不需要金鑰)
exchange = ccxt.binance()
ticker = exchange.fetch_ticker('BTC/USDT')
print(f"BTC 現價: {ticker['last']}")
print(f"24H 漲跌幅: {ticker['percentage']:.2f}%")
Step 4:啟用交易權限
⚠️ 重要安全提醒:
- 啟用「現貨交易」權限即可,不要啟用提幣權限
- 使用「子帳戶」隔離資金,限制單一機器人的最大風險
- 定期輪換 API Key(建議每 90 天)
🔒 API Key 安全警示(重要!)
警告:API Key 外洩可能導致資金被盜!
❌ 絕對禁止的做法
| 危險行為 | 風險說明 |
|----------|----------|
| 直接寫在程式碼中 | 上傳 GitHub 後被爬蟲掃描盜取 |
| 用 Email/Slack 傳送 | 通訊軟體可能被入侵 |
| 儲存在雲端文件 | Google Docs、Notion 可能被未授權存取 |
| 截圖分享 | 圖片可能被 OCR 辨識 |
✅ 正確的安全做法
#### 方法一:環境變數(推薦)
import os
from dotenv import load_dotenv
# 載入 .env 檔案
load_dotenv()
# 從環境變數讀取
API_KEY = os.getenv('BINANCE_API_KEY')
API_SECRET = os.getenv('BINANCE_API_SECRET')
# .env 檔案內容(記得加入 .gitignore!)
# BINANCE_API_KEY=your_key_here
# BINANCE_API_SECRET=your_secret_here
#### 方法二:金鑰管理服務
# AWS Secrets Manager 範例
import boto3
from botocore.exceptions import ClientError
def get_secret():
secret_name = "binance/api-credentials"
region_name = "us-east-1"
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
try:
secret_value = client.get_secret_value(SecretId=secret_name)
return json.loads(secret_value['SecretString'])
except ClientError as e:
print(f"無法取得金鑰: {e}")
return None
#### 方法三:硬體安全模組(HSM)
對於大資金帳戶,建議使用:
- AWS CloudHSM
- Azure Dedicated HSM
- 本地硬體錢包(如 Ledger)
安全檢查清單
- [ ] API Key 儲存在環境變數或金鑰管理服務
- [ ]
.env檔案已加入.gitignore - [ ] 已啟用 IP 白名單限制
- [ ] 已關閉提幣權限
- [ ] 使用子帳戶隔離資金
- [ ] 定期輪換 API Key
- [ ] 啟用交易所的登入通知與異常告警
風險控制與錯誤處理
為什麼風險控制比策略更重要?
華爾街名言:「風險控制是交易唯一免費的午餐。」
一個沒有風險控制的獲利策略,可能在一次黑天鵝事件中歸零。
必備的風險控制機制
#### 1. 停損機制
STOP_LOSS_PCT = 0.05 # 5% 停損
entry_price = None # 記錄進場價格
def check_stop_loss(current_price, entry_price):
"""
檢查是否觸發停損條件
參數:
current_price: 當前價格
entry_price: 進場價格
回傳:
True 如果觸發停損,False 否則
"""
if entry_price is None:
return False
loss_pct = (current_price - entry_price) / entry_price
if loss_pct < -STOP_LOSS_PCT:
print(f"🛑 觸發停損!虧損: {loss_pct:.2%}")
return True
return False
# 在交易邏輯中使用
if check_stop_loss(current_price, entry_price):
execute_trade(-1, SYMBOL, POSITION_SIZE) # 強制平倉
#### 2. 單筆部位限制
MAX_POSITION_USDT = 100 # 單筆最大 100 USDT
def validate_position_size(amount, current_price):
"""
驗證部位大小是否超過限制
參數:
amount: 交易數量
current_price: 當前價格
回傳:
(is_valid, message)
"""
position_value = amount * current_price
if position_value > MAX_POSITION_USDT:
return False, f"❌ 超過單筆部位上限 {MAX_POSITION_USDT} USDT"
return True, "✅ 部位大小驗證通過"
#### 3. 每日交易次數限制
MAX_TRADES_PER_DAY = 5
trade_count = 0
last_trade_date = None
def check_daily_trade_limit():
"""
檢查是否超過每日交易次數限制
回傳:
True 如果允許交易,False 如果已達上限
"""
global trade_count, last_trade_date
today = datetime.now().date()
if today != last_trade_date:
trade_count = 0
last_trade_date = today
if trade_count >= MAX_TRADES_PER_DAY:
print(f"❌ 已達今日交易上限 ({MAX_TRADES_PER_DAY} 次)")
return False
return True
def record_trade():
"""記錄一筆交易"""
global trade_count
trade_count += 1
print(f"📊 今日交易次數: {trade_count}/{MAX_TRADES_PER_DAY}")
#### 4. 異常監控與通知
import requests
def send_telegram_alert(message):
"""
發送 Telegram 警報通知
參數:
message: 警報訊息內容
設定方式:
1. 創建 Telegram Bot(透過 @BotFather)
2. 取得 Chat ID(透過 @userinfobot)
3. 將 Token 和 Chat ID 存入環境變數
"""
bot_token = os.getenv('TELEGRAM_BOT_TOKEN')
chat_id = os.getenv('TELEGRAM_CHAT_ID')
if not bot_token or not chat_id:
print("⚠️ Telegram 通知設定不完整")
return
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
payload = {
'chat_id': chat_id,
'text': message,
'parse_mode': 'HTML'
}
try:
response = requests.post(url, json=payload, timeout=10)
if response.status_code == 200:
print("✅ Telegram 通知已發送")
else:
print(f"❌ Telegram 通知失敗: {response.text}")
except Exception as e:
print(f"❌ Telegram 通知異常: {e}")
# 在錯誤處理中調用
except Exception as e:
send_telegram_alert(f"🚨 <b>交易機器人異常</b>\n\n錯誤: {e}\n時間: {datetime.now()}")
常見錯誤與解決方案
| 錯誤訊息 | 原因 | 解決方案 |
|----------|------|----------|
| Invalid API key | 金鑰錯誤或過期 | 重新生成 API Key |
| Insufficient balance | 餘額不足 | 檢查帳戶或降低下單量 |
| Rate limit exceeded | 請求過於頻繁 | 增加 sleep 時間或啟用 enableRateLimit |
| Network error | 網路連線問題 | 加入重試機制 |
| Invalid symbol | 交易對格式錯誤 | 確認格式為 'BTC/USDT' 而非 'BTCUSDT' |
Sentinel 免寫程式方案
覺得寫程式太麻煩?或者想要更專業的風險控制?
Sentinel 程式交易系統 提供零程式碼的完整解決方案:
Sentinel 核心功能
| 功能 | 說明 |
|------|------|
| 🎯 視覺化策略編輯器 | 拖拉積木就能建立策略,無需寫程式 |
| 📊 內建 50+ 技術指標 | RSI、MACD、布林通道、ATR 等一應俱全 |
| ⚡ 毫秒級執行 | 雲端部署,延遲低於 10ms |
| 🛡️ 多層風險控制 | 停損、止盈、最大回撤、資金管理全自動 |
| 📱 即時通知 | Telegram / Discord / Email 即時警報 |
| 📈 績效分析 | 自動生成夏普比率、最大回撤、勝率等報表 |
誰適合使用 Sentinel?
- ✅ 想專注策略開發,不想處理技術細節的交易者
- ✅ 需要同時管理多個策略、多個帳戶的專業玩家
- ✅ 對資金安全有極高要求的機構用戶
CTA:你的下一步
給程式派的你
- Fork 這份程式碼,在測試網(Testnet)上跑一週
- 加入停損機制,修改參數測試不同週期
- 嘗試其他指標,將雙均線改為 RSI 或 MACD
- 部署到雲端,使用 AWS Lambda 或 GCP 24/7 運行
給無程式派的你
- 註冊 Sentinel 試用帳號,體驗視覺化策略編輯
- 使用內建範本,快速啟動你的第一個自動化策略
- 連接模擬交易帳戶,零風險驗證策略有效性
相關閱讀
- 量化交易入門 2026|Python 自動交易策略完整指南(附 5 個範例程式碼)
- 交易機器人入門指南:7 步驟打造你的第一個自動交易策略(2026最新)
- 交易所 API 整合教學:Binance、OKX、Bybit 自動交易實戰完整指南|2026
- Webhook 入門教學:即時接收交易訊號與市場數據完整指南|2026
- 量化交易策略撰寫教學:從零開始的 Python 實戰範例(2026完整程式碼)