價格預測通常結合機器學習和深度學習演算法,並運用歷史數據及市場技術指標。以下是常見的一些方法:
使用強化學習的智能交易系統可以在模擬的市場環境中學習,並隨時間優化買賣決策,逐步改善交易策略。
AI 模型可以通過新聞和社交媒體的情感分析,來判斷市場情緒,並根據外部因素預測股票價格變化。
結合多種演算法和外部數據(例如經濟指標、公司基本面分析),通常能產生最有效的 AI 股票價格預測模型。
在使用 AI 進行股票價格預測時,模型會生成多種指標來幫助投資者做出決策。這些指標可以用來判斷市場走向,並根據不同的情境採取相應的交易操作。以下是一些常見的預測指標及其操作策略:
這是模型直接預測的未來股票價格數值。
模型可能會預測股票價格上漲或下跌的概率。
模型可能會基於技術指標(如移動平均線)預測股票的趨勢方向。
技術指標如 RSI(相對強弱指數)可以用來判斷股票是否處於超買或超賣狀態。
模型可以預測股票未來的價格波動區間。
基於市場買賣雙方的強度預測。
模型可能會根據歷史數據建議合理的止損與止盈點。
根據不同的預測指標,投資者可以做出相應的交易決策,如買入、賣出或保持觀望。綜合多種指標有助於在不同的市場情境下做出更合理的投資決策。
MetaQuotes Language(簡稱 MQL)是一種專為金融市場交易開發的程式語言,用於在 MetaTrader 平台(如 MT4 和 MT5)中創建自動交易策略(Expert Advisors)、自定指標、腳本及函數庫。
// 每次新 K 棒開始時進行操作
int start() {
if (OrdersTotal() == 0 && Volume[0] == 1) {
OrderSend(Symbol(), OP_BUY, 0.1, Ask, 3, 0, 0, "Buy Order", 0, 0, clrGreen);
}
return 0;
}
#property indicator_separate_window
#property indicator_buffers 1
double Buffer[];
int OnInit() {
SetIndexBuffer(0, Buffer);
return(INIT_SUCCEEDED);
}
int OnCalculate(const int rates_total, const int prev_calculated,
const datetime &time[], const double &open[],
const double &high[], const double &low[],
const double &close[], const long &tick_volume[],
const long &volume[], const int &spread[]) {
for (int i = 0; i < rates_total; i++) {
Buffer[i] = close[i] - open[i];
}
return(rates_total);
}
MQL 程式碼可透過 MetaTrader 內建的 MetaEditor 編輯與編譯,並在策略測試器中回測與優化。
MetaQuotes Language 是為交易自動化而生的專業語言,無論是初學者還是專業量化交易者,都能利用其強大功能實現精細交易策略。
Binance 是一個全球領先的加密貨幣交易所,為開發者提供豐富的 API 支援,包括 Spot API 和 Client API,方便使用者進行自動交易和數據獲取。
Binance Spot API 是 Binance 為現貨市場交易者設計的 API,可用於查詢市場資訊、下單、取消訂單等操作。此 API 常用於設計交易機器人、自動交易策略和監控市場波動。
Binance Client API 提供一個便捷的方式來存取 Binance 的各種 API 方法。開發者可以使用 binance.client 庫進行 API
認證和管理,並便捷地調用各種現貨和合約市場的功能。
pip install binancebinance.client 進行 API 連接,並調用 binance.spot 方法。
from binance.client import Client
# 初始化客戶端
client = Client(api_key='your_api_key', api_secret='your_secret_key')
# 取得當前的價格
price = client.get_symbol_ticker(symbol="BTCUSDT")
print(price)
Bybit 提供 REST 與 WebSocket API,可用於查詢行情、下單、查資金費率、管理帳戶等。以下展示如何使用 Python requests 套件呼叫 Bybit 公開 API。
pip install requests
import requests
BASE_URL = "https://api.bybit.com"
def get_symbols():
url = f"{BASE_URL}/v5/market/instruments-info?category=linear"
res = requests.get(url)
res.raise_for_status()
data = res.json()
symbols = [s["symbol"] for s in data["result"]["list"]]
print(f"共取得 {len(symbols)} 個交易對:")
print(symbols[:10]) # 顯示前 10 筆
if __name__ == "__main__":
get_symbols()
import requests
def get_klines(symbol="BTCUSDT", interval="60", limit=5):
url = f"{BASE_URL}/v5/market/kline?category=linear&symbol={symbol}&interval={interval}&limit={limit}"
res = requests.get(url)
res.raise_for_status()
data = res.json()
for k in data["result"]["list"]:
print(k)
if __name__ == "__main__":
get_klines()
Bybit 的下單、資產查詢等私有端點需要 API Key 與簽名。
import requests, time, hmac, hashlib
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"
def sign_request(params, secret):
"""Bybit 簽名生成"""
query = "&".join([f"{k}={v}" for k, v in sorted(params.items())])
return hmac.new(secret.encode(), query.encode(), hashlib.sha256).hexdigest()
def get_wallet_balance():
endpoint = "/v5/account/wallet-balance"
url = BASE_URL + endpoint
timestamp = str(int(time.time() * 1000))
params = {
"accountType": "UNIFIED",
"timestamp": timestamp,
"api_key": API_KEY,
}
params["sign"] = sign_request(params, API_SECRET)
res = requests.get(url, params=params)
print(res.json())
if __name__ == "__main__":
get_wallet_balance()
Pionex 使用 /api/v1/common/symbols?type=PERP 來取得「永續合約」交易對;在 Bybit,可使用 /v5/market/instruments-info 並指定 category=linear(USDT 永續)或 inverse(反向合約)達到同樣效果。
import requests
class BybitAPI:
BASE_URL = "https://api.bybit.com"
@classmethod
def get_symbols(cls, category="linear"):
"""
取得特定類型交易對
category 可為:
- linear → USDT 永續 (PERP)
- inverse → 反向永續/交割合約
- spot → 現貨
"""
endpoint = "/v5/market/instruments-info"
url = f"{cls.BASE_URL}{endpoint}"
params = {"category": category}
res = requests.get(url, params=params)
res.raise_for_status()
data = res.json()
if data.get("retCode") == 0:
symbols = [s["symbol"] for s in data["result"]["list"]]
print(f"共取得 {len(symbols)} 個 {category} 類型交易對")
for s in symbols[:10]:
print(s)
else:
print("取得失敗:", data)
if __name__ == "__main__":
BybitAPI.get_symbols("linear")
| Pionex | Bybit | 說明 |
|---|---|---|
type=PERP | category=linear | USDT 永續合約 |
type=SPOT | category=spot | 現貨市場 |
| — | category=inverse | 反向永續或交割合約 |
{
"retCode": 0,
"result": {
"list": [
{
"symbol": "BTCUSDT",
"contractType": "LinearPerpetual",
"status": "Trading",
"lotSizeFilter": {
"minOrderQty": "0.001",
"maxOrderQty": "100",
"qtyStep": "0.001"
},
"priceFilter": {
"tickSize": "0.5"
}
}
]
}
}
category=linear 即對應到 Pionex 的「PERP」交易對。priceFilter、lotSizeFilter 等限制。在 Pionex 中可使用 /api/v1/market/klines 查行情;Bybit 對應的端點是 /v5/market/kline。透過 category 指定市場類型(例如 linear 表示 USDT 永續合約),並可傳入 symbol、interval、limit、endTime 等參數。
import requests
import time
class BybitAPI:
BASE_URL = "https://api.bybit.com"
@classmethod
def get_klines(cls, symbol: str, interval: str, end_time: int = None, limit: int = 100):
"""
查詢 Bybit K線資料
:param symbol: 交易對,例如 "BTCUSDT"
:param interval: 時間區間 (1, 3, 5, 15, 30, 60, 120, 240, 360, 720, D, W, M)
:param end_time: 結束時間 (Unix 毫秒),預設為現在
:param limit: 回傳筆數,最大 1000
"""
endpoint = "/v5/market/kline"
url = f"{cls.BASE_URL}{endpoint}"
params = {
"category": "linear", # USDT 永續
"symbol": symbol,
"interval": interval,
"limit": limit
}
if end_time:
params["end"] = end_time
else:
params["end"] = int(time.time() * 1000)
res = requests.get(url, params=params)
res.raise_for_status()
data = res.json()
if data.get("retCode") == 0:
klines = data["result"]["list"]
print(f"{symbol} 共取得 {len(klines)} 根 K 線")
# 顯示前 3 根資料
for k in klines[:3]:
open_time, open_price, high, low, close, volume, turnover = k
print(f"開:{open_price} 收:{close} 高:{high} 低:{low} 量:{volume}")
else:
print("取得失敗:", data)
if __name__ == "__main__":
BybitAPI.get_klines(symbol="BTCUSDT", interval="60", limit=5)
{
"retCode": 0,
"result": {
"symbol": "BTCUSDT",
"category": "linear",
"list": [
[
"1735119600000", // 開始時間 (毫秒)
"98342.5", // 開盤價
"98350.0", // 最高價
"98285.0", // 最低價
"98290.5", // 收盤價
"12.304", // 交易量
"1210000.5" // 成交額 (USDT)
]
]
}
}
category 可改為 spot 或 inverse 對應不同市場。end 參數為毫秒時間戳(Unix epoch × 1000)。下面是一個可直接複製的 Python 範例,採用 ThreadPoolExecutor 並內建重試、速率限制與快取(cache)。流程:先從快取載入已抓到的 K 線(若有),只對缺少或不足的交易對發出請求;使用 thread pool 同時執行多請求並以 semaphore 控制並發數避免被限流;遇到失敗會 exponential backoff 重試。
# 需求: pip install requests
import requests
import time
import json
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Semaphore
BASE_URL = "https://api.bybit.com/v5/market/kline"
CACHE_FILE = "klines_cache.json"
# 可調參數
MAX_WORKERS = 20 # thread pool 大小(視 API 速率限制調整)
MAX_CONCURRENT = 10 # 真正並發的請求數(用 semaphore 控制)
RETRY = 3 # 每個請求的重試次數
INITIAL_BACKOFF = 0.5 # 首次重試等待秒數
CATEGORY = "linear" # linear -> USDT 永續;spot/inverse 可改
LIMIT_PER_CALL = 200 # 每次 kline API limit(視 API 上限設定)
DATASET_ALLDAYS = 24 * 6 # 範例:判定需要至少多少根 K 線(可改)
# 載入/存取快取
def load_cache():
if os.path.exists(CACHE_FILE):
try:
return json.load(open(CACHE_FILE, "r", encoding="utf-8"))
except Exception:
return {}
return {}
def save_cache(cache):
json.dump(cache, open(CACHE_FILE, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
# 單一 symbol 的 API 抓取,含重試與速率控制(由 semaphore 控制同時執行數)
def fetch_klines_for_symbol(symbol, interval="60", end_time=None, limit=LIMIT_PER_CALL, sem: Semaphore = None):
params = {
"category": CATEGORY,
"symbol": symbol,
"interval": interval,
"limit": limit
}
if end_time:
params["end"] = int(end_time)
backoff = INITIAL_BACKOFF
last_exc = None
# 取得 semaphore(若提供)
if sem:
sem.acquire()
try:
for attempt in range(1, RETRY + 1):
try:
resp = requests.get(BASE_URL, params=params, timeout=10)
resp.raise_for_status()
data = resp.json()
# Bybit v5 回傳 retCode == 0 才表示成功
if data.get("retCode", 0) == 0 and "result" in data:
klines = data["result"].get("list", [])
return klines
else:
last_exc = Exception(f"API error: {data}")
except requests.exceptions.RequestException as e:
last_exc = e
# backoff
time.sleep(backoff)
backoff *= 2
finally:
if sem:
sem.release()
# 若失敗,拋出最後一個錯誤或回傳 None
raise last_exc
# 批次處理:傳入 pairs(list of symbols),回傳 dict {symbol: klines}
def get_klines_batch(pairs, interval="60", dataset_alldays=DATASET_ALLDAYS, limit=LIMIT_PER_CALL, max_workers=MAX_WORKERS, max_concurrent=MAX_CONCURRENT):
cache = load_cache() # cache 格式: { symbol: [kline_list] }
results = {}
to_fetch = []
# 判斷哪些 symbol 需要抓(cache 中不存在或長度不足)
for s in pairs:
cached = cache.get(s)
if cached and len(cached) >= dataset_alldays:
results[s] = cached
else:
to_fetch.append(s)
# 如果沒有需要抓取的就直接回傳
if not to_fetch:
return results
sem = Semaphore(max_concurrent)
with ThreadPoolExecutor(max_workers=max_workers) as exe:
futures = {exe.submit(fetch_klines_for_symbol, sym, interval, None, limit, sem): sym for sym in to_fetch}
for fut in as_completed(futures):
sym = futures[fut]
try:
kl = fut.result()
# 若 API 回傳以列表格式儲存(依 Bybit 範例 [time,open,high,low,close,vol,turnover])
cache[sym] = kl
results[sym] = kl
except Exception as e:
# 記錄失敗,但不阻斷整體程序
print(f"[錯誤] {sym} 取得失敗: {e}")
results[sym] = None
# 將 cache 存檔(可選:只存成功的)
save_cache(cache)
return results
# 範例使用
if __name__ == "__main__":
# 假設有 500 個 pairs(示意)
pairs = ["BTCUSDT", "ETHUSDT", "SOLUSDT"] # ... 500 個
# 執行批次抓取
all_klines = get_klines_batch(pairs, interval="60", dataset_alldays=100, limit=200)
# 篩選符合長度需求的 symbols
good = [s for s, kl in all_klines.items() if kl and len(kl) >= 100]
print(f"符合 100 根以上的交易對數量: {len(good)}")
print(good[:20])
MAX_WORKERS 與 MAX_CONCURRENT。若有 API key,通常能提高速率配額。Pionex 提供官方 API,讓開發者能透過程式自動化交易、查詢市場數據、管理帳戶資產。API 支援 REST 與 WebSocket 兩種方式。
API Key 與 Secret,僅顯示一次
// 使用 Node.js axios 請求 Pionex API
const axios = require("axios");
const crypto = require("crypto");
const apiKey = "你的API_KEY";
const secret = "你的API_SECRET";
const baseUrl = "https://api.pionex.com";
// 簽名產生
function sign(query) {
return crypto.createHmac("sha256", secret).update(query).digest("hex");
}
// 查詢帳戶餘額
async function getBalances() {
const timestamp = Date.now();
const query = `timestamp=${timestamp}`;
const signature = sign(query);
const res = await axios.get(`${baseUrl}/api/v1/account?${query}&signature=${signature}`, {
headers: { "X-MBX-APIKEY": apiKey }
});
console.log(res.data);
}
getBalances();
import time
import hmac
import hashlib
import requests
API_KEY = "你的API_KEY"
SECRET = "你的API_SECRET"
BASE_URL = "https://api.pionex.com"
def sign(query: str) -> str:
return hmac.new(SECRET.encode(), query.encode(), hashlib.sha256).hexdigest()
def get_balances():
timestamp = str(int(time.time() * 1000))
query = f"timestamp={timestamp}"
signature = sign(query)
url = f"{BASE_URL}/api/v1/account?{query}&signature={signature}"
headers = {"X-MBX-APIKEY": API_KEY}
res = requests.get(url, headers=headers)
print(res.json())
get_balances()
const WebSocket = require("ws");
const ws = new WebSocket("wss://ws.pionex.com/ws");
ws.on("open", () => {
console.log("已連線至 Pionex WebSocket");
// 訂閱 BTC/USDT 行情
ws.send(JSON.stringify({
event: "subscribe",
channel: "market",
market: "BTC_USDT"
}));
});
ws.on("message", (msg) => {
console.log("接收訊息:", msg.toString());
});
import websocket
import json
def on_open(ws):
print("已連線至 Pionex WebSocket")
sub_msg = {
"event": "subscribe",
"channel": "market",
"market": "BTC_USDT"
}
ws.send(json.dumps(sub_msg))
def on_message(ws, message):
print("接收訊息:", message)
ws = websocket.WebSocketApp(
"wss://ws.pionex.com/ws",
on_open=on_open,
on_message=on_message
)
ws.run_forever()
可使用 GET /api/v1/common/symbols 來取得 Pionex 所有支援的交易對與詳細屬性,例如最小下單量、價格精度、交易類型(現貨或合約)等。
GET https://api.pionex.com/api/v1/common/symbols
{
"code": 0,
"data": [
{
"symbol": "BTC_USDT",
"quoteCurrency": "USDT",
"baseCurrency": "BTC",
"minQty": "0.0001",
"minNotional": "5",
"pricePrecision": 2,
"quantityPrecision": 6,
"tradeEnable": true
},
{
"symbol": "ETH_USDT",
"quoteCurrency": "USDT",
"baseCurrency": "ETH",
"minQty": "0.001",
"pricePrecision": 2,
"quantityPrecision": 6,
"tradeEnable": true
}
]
}
import requests
BASE_URL = "https://api.pionex.com"
def get_symbols():
url = f"{BASE_URL}/api/v1/common/symbols"
res = requests.get(url)
data = res.json()
if data.get("code") == 0:
symbols = data.get("data", [])
print(f"共取得 {len(symbols)} 個交易對")
for s in symbols[:10]: # 只顯示前10個
print(f"{s['symbol']} ({s['baseCurrency']}/{s['quoteCurrency']})")
else:
print("取得失敗:", data)
if __name__ == "__main__":
get_symbols()
若要只取出支援永續合約的交易對,可使用下列簡單篩選邏輯:
perp_symbols = [s for s in data["data"] if ".PERP" in s["symbol"]]
/api/v1/market/tickers 取得即時價格。以下程式會呼叫 https://api.pionex.com/api/v1/common/symbols,自動印出回傳 JSON 的結構(key 與資料型態),方便了解實際格式。
import requests
import json
def print_json_structure(data, indent=0):
"""遞迴印出 JSON 結構"""
space = " " * indent
if isinstance(data, dict):
for k, v in data.items():
if isinstance(v, (dict, list)):
print(f"{space}{k}: {type(v).__name__}")
print_json_structure(v, indent + 1)
else:
print(f"{space}{k}: {type(v).__name__}")
elif isinstance(data, list) and data:
print(f"{space}[list] item type: {type(data[0]).__name__}")
print_json_structure(data[0], indent + 1)
def get_pionex_symbols_format():
url = "https://api.pionex.com/api/v1/common/symbols"
res = requests.get(url)
res.raise_for_status()
data = res.json()
print("根層結構:")
print_json_structure(data)
if __name__ == "__main__":
get_pionex_symbols_format()
根層結構:
code: int
data: list
[list] item type: dict
symbol: str
baseCurrency: str
quoteCurrency: str
pricePrecision: int
quantityPrecision: int
minQty: str
minNotional: str
tradeEnable: bool
json.dumps(data, indent=2) 可檢視完整內容。取得指定交易對的 K 線(Candlestick / OHLCV)數據,來源來自 Pionex 公開市場資料。
| 參數 | 型別 | 是否必需 | 說明 |
|---|---|---|---|
| symbol | string | 是 | 交易對 (例:BTC_USDT 或 BTC_USDT.PERP) |
| interval | string | 是 | 時間間隔,例如 1M、5M、15M、30M、60M、4H、8H、12H、1D |
| endTime | number (毫秒) | 否 | 結束時間(毫秒時間戳) |
| limit | number | 否 | 取得資料數量,預設 100,範圍 1-500 |
{
"result": true,
"data": {
"klines": [
{
"time": 1691649240000,
"open": "1851.27",
"close": "1851.32",
"high": "1851.32",
"low": "1851.27",
"volume": "0.542"
}
]
},
"timestamp": 1691649271544
}
import requests
BASE_URL = "https://api.pionex.com"
def get_klines(symbol: str, interval: str, end_time: int = None, limit: int = 100):
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
if end_time is not None:
params["endTime"] = end_time
response = requests.get(f"{BASE_URL}/api/v1/market/klines", params=params)
result = response.json()
if result.get("result") and "data" in result:
return result["data"]["klines"]
else:
raise Exception(f"取得 K 線失敗: {result}")
if __name__ == "__main__":
# 範例:取得 BTC_USDT 永續合約最近 50 根 15 分鐘 K 線
symbol = "BTC_USDT.PERP"
interval = "15M"
klines = get_klines(symbol, interval, limit=50)
for k in klines:
print(k)
Pionex 並沒有單獨一個 「取得支援網格交易對」 的專用 API,但可以透過 GET /api/v1/market/tickers 取得所有交易對,再從中篩選出 .PERP 類型(USDT 永續合約),即可得到 Futures Grid 支援的交易對清單。
import requests
BASE_URL = "https://api.pionex.com"
def get_perp_pairs():
url = f"{BASE_URL}/api/v1/market/tickers"
res = requests.get(url)
data = res.json()
perp_pairs = []
if "data" in data:
for item in data["data"]:
market = item.get("symbol", "")
# 永續合約交易對通常以 .PERP 結尾
if ".PERP" in market:
perp_pairs.append(market)
return perp_pairs
if __name__ == "__main__":
pairs = get_perp_pairs()
print("支援的永續合約網格交易對:")
for p in pairs:
print(p)
支援的永續合約網格交易對:
BTC_USDT.PERP
ETH_USDT.PERP
SOL_USDT.PERP
LINK_USDT.PERP
...
.PERP)可直接用於 Futures Grid Bot。Max Coin API 是由 Max Exchange 提供的一套應用程式介面,允許開發者程式化地訪問其加密貨幣交易功能。開發者可以透過 API 自動化執行交易、檢索市場數據以及管理資產。
以下是一個透過 API 獲取市場數據的範例:
GET https://max-api.maicoin.com/api/v1/ticker?market=btctwd
該請求將返回 BTC/TWD 的即時市場數據,包括價格、成交量等。
以下是一個使用 Python 語言調用 Max API 的簡單範例:
import requests
BASE_URL = "https://max-api.maicoin.com"
def get_ticker(pair):
endpoint = "/api/v1/ticker"
params = {"market": pair}
response = requests.get(BASE_URL + endpoint, params=params)
return response.json()
# 獲取 BTC/TWD 市場數據
ticker_data = get_ticker("btctwd")
print(ticker_data)
比特幣使用 工作量證明(Proof of Work, PoW) 作為共識機制,以實現去中心化的記帳系統。PoW 的核心目標是讓節點透過競爭解數學難題來決定誰擁有記帳權,確保資料無法被任意竄改。
礦工會根據以下條件優先挑選交易,以提高成功打包區塊並賺取手續費的機會:
當交易量過大時,出現擁堵與手續費上升的情況。為改善此問題,提出多種擴展解決方案:
| 欄位名稱 | 大小(位元組) | 說明 |
|---|---|---|
| block size | 4 | 整個區塊(含 header 與所有交易資料)的總大小(以 byte 為單位) |
| block header | 80 | 區塊的標頭資訊,用來做驗證與連結前後區塊 |
| transaction counter | 1 ~ 9 | 交易數量,以變長整數(VarInt)表示本區塊內有幾筆交易 |
| transactions | 可變 | 所有實際的交易資料,每一筆交易包含輸入與輸出 |
長度固定為 80 個位元組,內容如下:
| 欄位名稱 | 資料型別 | 長度 | 說明 |
|---|---|---|---|
| version | int32 | 4 | 區塊版本,代表可接受的區塊驗證規則 |
| previous block hash | char[32] | 32 | 前一個區塊的哈希值 |
| merkle root | char[32] | 32 | 所有交易哈希值的 Merkle Tree 根 |
| timestamp | uint32 | 4 | 區塊建立時間(UNIX 時間戳) |
| bits | uint32 | 4 | 目標難度的壓縮表示 |
| nonce | uint32 | 4 | 為了找到合法區塊哈希的變動值 |
使用 VarInt(變長整數)格式,表示區塊中有多少筆交易:
每一筆交易資料包含以下主要部分(長度可變):
第一筆交易通常是 coinbase 交易,也就是礦工領取區塊獎勵的特殊交易,不包含輸入。
import time
class Transaction:
def __init__(self, sender, receiver, amount, fee):
self.sender = sender
self.receiver = receiver
self.amount = amount
self.fee = fee
self.timestamp = time.time()
def __repr__(self):
return f"[Tx: {self.sender} → {self.receiver}, ${self.amount}, fee: {self.fee}]"
class Node:
def __init__(self, name):
self.name = name
self.peers = []
self.transaction_pool = []
def connect(self, peer):
if peer not in self.peers:
self.peers.append(peer)
peer.connect(self) # 雙向連線
def receive_transaction(self, tx):
if tx not in self.transaction_pool:
self.transaction_pool.append(tx)
print(f"{self.name} 收到交易:{tx}")
self.broadcast(tx)
def broadcast(self, tx):
for peer in self.peers:
peer.receive_transaction(tx)
class Miner(Node):
def mine_block(self):
print(f"\n⛏️ {self.name} 開始打包區塊")
# 按手續費高排序,取最多 5 筆交易
sorted_txs = sorted(self.transaction_pool, key=lambda tx: tx.fee, reverse=True)
selected = sorted_txs[:5]
print(f"{self.name} 打包交易:")
for tx in selected:
print(f" - {tx}")
# 清空已處理的交易
self.transaction_pool = [tx for tx in self.transaction_pool if tx not in selected]
# 建立節點與礦工
A = Node("節點 A")
B = Node("節點 B")
C = Miner("礦工 C")
# 連接節點網路
A.connect(B)
B.connect(C)
# 用戶發出交易
tx_list = [
Transaction("Alice", "Bob", 2.0, 0.0005),
Transaction("Eve", "Tom", 1.2, 0.0009),
Transaction("Joe", "Mary", 3.5, 0.0002),
Transaction("Rick", "Sam", 0.8, 0.0015),
Transaction("Ann", "Lily", 1.7, 0.0001)
]
for tx in tx_list:
print(f"\n用戶送出交易:{tx}")
A.receive_transaction(tx)
time.sleep(0.2)
# 礦工開始打包
C.mine_block()
Ethereum 虛擬機(Ethereum Virtual Machine,簡稱 EVM)是以太坊的核心組件,負責執行智能合約。EVM 提供了一個沙箱環境,允許開發者在上面運行代碼,而不需要擔心會影響到以太坊網絡的其他部分。
智能合約是自動執行、不可更改的合約,執行時由 EVM 處理。開發者通常使用 Solidity 等高級編程語言來撰寫智能合約,然後將其編譯成 EVM 可理解的字節碼。
Solidity 是 Ethereum 上最常用的編程語言,語法類似於 JavaScript。以下是一個簡單的 Solidity 智能合約示例:
pragma solidity ^0.8.0;
contract SimpleStorage {
uint public storedData;
function set(uint x) public {
storedData = x;
}
function get() public view returns (uint) {
return storedData;
}
}
上述合約包含一個用來儲存整數數據的變量 storedData,以及設置和獲取該數據的函數。
當用戶在以太坊網絡上執行一個智能合約時,以下步驟會發生:
EVM 的運算資源是有限的,為了防止網絡濫用,EVM 使用 Gas 機制來計算和收取交易費用。每個操作都有其相應的 Gas 成本,用戶在提交交易時需要提供足夠的 Gas
來支付執行智能合約所需的計算資源。
EVM 是以太坊網絡的核心,提供了一個強大的環境來運行智能合約。通過使用 Solidity 等編程語言,開發者可以創建各種去中心化應用(dApps),並利用 EVM 的功能來實現複雜的邏輯運算和交易處理。
Node.js 和 npmHardhat:npm install --save-dev hardhatnpx hardhat,選擇「Create a basic sample project」在 contracts 資料夾中新增 LendingProtocol.sol 並貼上你的 Solidity 合約。
於 scripts 資料夾建立 deploy.js,內容如下:
async function main() {
const [deployer] = await ethers.getSigners();
console.log("部署帳號:", deployer.address);
const TokenAddress = "0xYourTokenAddressHere";
const LendingProtocol = await ethers.getContractFactory("LendingProtocol");
const lending = await LendingProtocol.deploy(TokenAddress);
await lending.deployed();
console.log("LendingProtocol 部署成功:", lending.address);
}
main().catch((error) => {
console.error(error);
process.exitCode = 1;
});
啟動 Hardhat 測試鏈:
npx hardhat node
另開一個 terminal 部署:
npx hardhat run scripts/deploy.js --network localhost
hardhat.config.js 中加入:
require("@nomiclabs/hardhat-ethers");
module.exports = {
networks: {
goerli: {
url: "https://goerli.infura.io/v3/你的API金鑰",
accounts: ["0x你的私鑰"]
}
},
solidity: "0.8.20"
};
然後部署:
npx hardhat run scripts/deploy.js --network goerli
部署後會輸出合約地址,可用於前端整合與互動。
本借貸協議為以太坊虛擬機(EVM)上運行的智慧合約,用戶可以透過本合約存入資產賺取利息或借出資產支付利息。此協議支援ERC-20代幣,具備借貸、儲蓄、清算等核心功能。
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;
interface IERC20 {
function transferFrom(address sender, address recipient, uint amount) external returns (bool);
function transfer(address recipient, uint amount) external returns (bool);
function balanceOf(address account) external view returns (uint);
function approve(address spender, uint amount) external returns (bool);
}
contract LendingProtocol {
IERC20 public token;
address public owner;
uint public interestRate = 5; // 年利率5%
mapping(address => uint) public deposits;
mapping(address => uint) public borrows;
constructor(address _token) {
token = IERC20(_token);
owner = msg.sender;
}
function deposit(uint amount) external {
require(amount > 0, "金額需大於0");
token.transferFrom(msg.sender, address(this), amount);
deposits[msg.sender] += amount;
}
function borrow(uint amount) external {
require(amount > 0, "金額需大於0");
uint collateral = deposits[msg.sender];
require(collateral >= amount * 2, "抵押不足");
borrows[msg.sender] += amount;
token.transfer(msg.sender, amount);
}
function repay(uint amount) external {
require(amount > 0, "金額需大於0");
require(borrows[msg.sender] >= amount, "借款不足");
borrows[msg.sender] -= amount;
token.transferFrom(msg.sender, address(this), amount);
}
function withdraw(uint amount) external {
require(deposits[msg.sender] >= amount, "餘額不足");
require(borrows[msg.sender] == 0, "有未還借款");
deposits[msg.sender] -= amount;
token.transfer(msg.sender, amount);
}
}
BSC (Binance Smart Chain) 是一條相容於 EVM (Ethereum Virtual Machine) 的公鏈,因此在以太坊上編寫的 Solidity 智能合約幾乎可以直接部署到 BSC。主要差異在於部署時連接的網路與 Gas 費以 BNB 計價。
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;
contract SimpleStorage {
uint private value;
function setValue(uint _value) public {
value = _value;
}
function getValue() public view returns (uint) {
return value;
}
}
使用 Hardhat 進行部署,在 hardhat.config.js 中設定 BSC 網路:
require("@nomiclabs/hardhat-ethers");
module.exports = {
solidity: "0.8.20",
networks: {
bscTestnet: {
url: "https://data-seed-prebsc-1-s1.binance.org:8545/",
chainId: 97,
gasPrice: 20000000000,
accounts: ["0x你的私鑰"]
},
bscMainnet: {
url: "https://bsc-dataseed.binance.org/",
chainId: 56,
gasPrice: 20000000000,
accounts: ["0x你的私鑰"]
}
}
};
async function main() {
const [deployer] = await ethers.getSigners();
console.log("部署帳號:", deployer.address);
const SimpleStorage = await ethers.getContractFactory("SimpleStorage");
const storage = await SimpleStorage.deploy();
await storage.deployed();
console.log("合約已部署:", storage.address);
}
main().catch((error) => {
console.error(error);
process.exitCode = 1;
});
npm install --save-dev hardhat @nomiclabs/hardhat-ethers ethersnpx hardhat run scripts/deploy.js --network bscTestnet使用 Python 和隨機森林模型來預測股票的漲跌概率。此範例利用 Yahoo 財經的股票數據,並通過技術指標來訓練模型,最後輸出股票上漲和下跌的概率。
首先,我們需要安裝並匯入 Python 的一些套件:
pip install yfinance scikit-learn pandas numpy
接著匯入這些必要的庫:
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
import yfinance as yf
接下來,我們從 Yahoo 財經下載股票的歷史數據,並計算簡單移動平均線 (SMA) 作為技術指標。
# 從 Yahoo 財經下載蘋果公司的股票數據
symbol = 'AAPL'
data = yf.download(symbol, start='2020-01-01', end='2023-01-01')
# 計算 10 日和 50 日簡單移動平均線 (SMA)
data['SMA_10'] = data['Close'].rolling(window=10).mean()
data['SMA_50'] = data['Close'].rolling(window=50).mean()
# 設置漲跌目標,若次日收盤價高於當日收盤價,則為1 (表示上漲),否則為0 (表示下跌)
data['Target'] = np.where(data['Close'].shift(-1) > data['Close'], 1, 0)
# 移除缺失值
data.dropna(inplace=True)
我們將使用移動平均線 (SMA) 和收盤價作為特徵,並將數據分為訓練集和測試集。
# 選擇特徵
features = ['SMA_10', 'SMA_50', 'Close']
X = data[features]
y = data['Target']
# 分割訓練集與測試集 (80%訓練,20%測試)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False)
我們使用隨機森林模型來訓練數據,並預測測試集中股票的漲跌概率。
# 初始化隨機森林分類器
model = RandomForestClassifier(n_estimators=100, random_state=42)
# 訓練模型
model.fit(X_train, y_train)
# 預測測試集中的漲跌
y_pred = model.predict(X_test)
y_prob = model.predict_proba(X_test) # 獲取漲跌的概率
我們可以計算模型的準確率,並顯示每一天的漲跌概率。
# 計算模型準確率
accuracy = accuracy_score(y_test, y_pred)
print(f"模型準確率: {accuracy:.2f}")
# 顯示測試集中前 5 天的漲跌概率
for i in range(5):
print(f"第{i+1}天:上漲概率={y_prob[i][1]:.2f}, 下跌概率={y_prob[i][0]:.2f}")
根據上漲的概率,您可以設定一個閾值來決定是否進行買入操作。例如,如果上漲概率超過 70%,則買入。
# 設定上漲概率的閾值
threshold = 0.7
# 根據概率進行買入決策
for i in range(len(y_prob)):
if y_prob[i][1] > threshold:
print(f"第{i+1}天建議買入,預測上漲概率={y_prob[i][1]:.2f}")
else:
print(f"第{i+1}天不建議買入,預測上漲概率={y_prob[i][1]:.2f}")
這個範例展示了如何使用隨機森林模型來預測股票的漲跌概率,並根據預測結果進行交易決策。這是一種簡單但有效的方式來提高交易決策的準確性。
可以透過以下網址訪問台灣證券交易所的公開申購公告頁面:
使用 Python 搭配 requests 和 BeautifulSoup,即可抓取頁面上的公開申購資料。
import requests
from bs4 import BeautifulSoup
import pandas as pd
# 抓取公開申購公告頁面
url = "https://www.twse.com.tw/zh/announcement/public.html"
headers = {"User-Agent": "Mozilla/5.0"}
response = requests.get(url, headers=headers)
if response.status_code == 200:
soup = BeautifulSoup(response.text, 'html.parser')
tables = pd.read_html(response.text)
if tables:
df = tables[0] # 取第一個表格
print(df)
else:
print("未找到表格資料")
else:
print("無法連接到公開申購公告頁面")
Selenium 模擬瀏覽器操作。CurrencyAPI 提供即時匯率資訊。以下是範例程式碼:
import requests
url = 'https://api.currencyapi.com/v3/latest'
params = {
'apikey': '您的API金鑰',
'base_currency': 'USD',
'currencies': 'TWD'
}
response = requests.get(url, params=params)
data = response.json()
usd_to_twd = data['data']['TWD']['value']
print(f"1 美元等於 {usd_to_twd} 新台幣")
注意:需至 CurrencyAPI 註冊並取得 API 金鑰後才能使用。
ExchangeRatesAPI 同樣提供即時匯率查詢服務:
import requests
url = 'https://api.exchangeratesapi.io/latest'
params = {
'access_key': '您的API金鑰',
'base': 'USD',
'symbols': 'TWD'
}
response = requests.get(url, params=params)
data = response.json()
usd_to_twd = data['rates']['TWD']
print(f"1 美元等於 {usd_to_twd} 新台幣")
請先至 ExchangeRatesAPI 註冊取得 API 金鑰並替換程式碼中的 '您的API金鑰'。
若不想直接調用 API,可使用第三方 Python 套件 forex-python:
from forex_python.converter import CurrencyRates
cr = CurrencyRates()
usd_to_twd = cr.get_rate('USD', 'TWD')
print(f"1 美元等於 {usd_to_twd} 新台幣")
安裝套件命令:
pip install forex-python
若不使用 API,可以透過網頁爬蟲技術直接從 Currency.Wiki 網站擷取美元對台幣的即時匯率。
需要安裝以下 Python 套件:
pip install requests pip install beautifulsoup4
import requests
from bs4 import BeautifulSoup
# 設定目標網址
url = "https://currency.wiki/usd_twd"
# 發送 GET 請求取得網頁內容
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
# 查找特定標籤和類別
span_tag = soup.find('span', class_='unit_secondary_value')
rate = span_tag.text
print(f"1 美元等於 {rate} 新台幣")
透過 Python 爬蟲技術可直接取得即時匯率資訊,但需注意網頁結構變動和合規性問題,適合小規模應用或學習用途。
email: [email protected]