巴基斯坦證券交易所（PSX）股票API對接：即時行情與量化交易實踐指南

      import os
from dotenv import load_dotenv
load_dotenv()

ITICK_API_KEY = os.getenv("ITICK_API_KEY", "your_key_here")

    

      import requests
import pandas as pd
from typing import Optional, List, Dict
import os

def fetch_psx_symbols(api_key: str) -> Optional[List[Dict]]:
    url = "https://api.itick.org/symbol/list?type=stock&region=PK"
    headers = {
        "accept": "application/json",
        "token": api_key  # iTick使用token頭進行認證
    }
    try:
        print(f"🔄 正在請求PSX股票列表...")
        response = requests.get(
            url, 
            headers=headers, 
            timeout=15
        )
        
        # 處理HTTP狀態碼
        if response.status_code == 200:
            data = response.json()
            if data.get("code") == 200:
                symbol_list = data.get("data", [])
                print(f"✅ 成功獲取 {len(symbol_list)} 只PSX股票")
                return symbol_list
            else:
                print(f"⚠️ API業務錯誤: {data.get('msg')} (代碼: {data.get('code')})")
                return None
                
        elif response.status_code == 401:
            print("❌ 認證失敗 (401): 請檢查API密鑰是否有效、過期，或該接口未被您的套餐覆蓋")
            print("   - 登錄 https://itick.org 控制台查看密鑰狀態")
            print("   - 確認您的套餐是否包含'股票基礎數據'權限")
            return None
            
        elif response.status_code == 403:
            print("❌ 權限不足 (403): 您的帳號無權訪問PSX區域數據")
            print("   - PSX目前可能是受限測試狀態，需聯繫商務開通")
            return None
            
        elif response.status_code == 429:
            print("❌ 請求過於頻繁 (429): 請稍後重試，或升級套餐提高速率限制")
            return None
            
        else:
            print(f"❌ 未知錯誤: HTTP {response.status_code}")
            return None
    except Exception as e:
        print(f"❌ 請求異常: {str(e)}")
        return None


def save_to_dataframe(symbol_list: List[Dict]) -> pd.DataFrame:
    """將股票列表保存為Pandas DataFrame便於分析"""
    if not symbol_list:
        return pd.DataFrame()
    
    df = pd.DataFrame(symbol_list)
    print("\n📋 PSX股票列表預覽:")
    print(df.head())
    print(f"\n總記錄數: {len(df)}")
    return df


# ===== 使用示例 =====
if __name__ == "__main__":
    # ⚠️ 重要: 請從環境變量或配置文件中讀取密鑰，切勿硬編碼
    YOUR_API_KEY = os.environ.get("ITICK_API_KEY", "")
    
    if not YOUR_API_KEY:
        print("請先設置環境變量 ITICK_API_KEY，或直接賦值（僅測試）")
        # YOUR_API_KEY = "your_key_here"  # 測試時可臨時取消註釋
    else:
        # 步驟1: 獲取PSX股票列表
        symbols = fetch_psx_symbols(YOUR_API_KEY)
        
        # 步驟2: 轉換為DataFrame
        if symbols:
            df = save_to_dataframe(symbols)
            
            # 可選: 保存到CSV
            df.to_csv("psx_symbols.csv", index=False)
            print("\n💾 已保存至 psx_symbols.csv")
            
            # 打印所有股票代碼
            print("\n🔹 PSX股票代碼列表:")
            for code in df.get("symbol", [])[:20]:  # 顯示前20個
                print(f"  {code}")

    

      import requests
import pandas as pd
from datetime import datetime, timedelta

class PSXDataFetcher:
    """巴基斯坦PSX市場數據獲取器（基於iTick API協議適配）"""
    
    BASE_URL = "https://api.itick.org/stock/kline"
    
    def __init__(self, api_key):
        self.headers = {
            "accept": "application/json",
            "token": api_key
        }
    
    def get_historical_data(self, symbol, ktype=5, limit=100):
        """
        獲取PSX股票歷史K線數據
        :param symbol: PSX股票代碼（如'OGDC'）
        :param ktype: K線週期（1=1分鐘, 2=5分鐘, 3=15分鐘, 4=30分鐘, 5=1小時, 6=2小時, 7=4小時, 8=日線, 9=週線, 10=月線）
        :param limit: 返回K線數量（最大500）
        :return: Pandas DataFrame
        """
        # PSX市場通過自定義region參數適配
        params = {
            "region": "PK",      # 市場標識
            "code": symbol,
            "kType": ktype,
            "limit": limit
        }
        
        try:
            response = requests.get(
                self.BASE_URL, 
                headers=self.headers, 
                params=params,
                timeout=10
            )
            response.raise_for_status()
            data = response.json()
            
            if data.get("code") == 200:
                return self._parse_kline(data.get("data", []))
            else:
                print(f"API錯誤: {data.get('msg')}")
                return None
                
        except Exception as e:
            print(f"請求失敗 {symbol}: {str(e)}")
            # 模擬數據生成（用於演示，實際使用時應刪除）
            return self._generate_mock_data(symbol, limit)
    
    def _parse_kline(self, kline_list):
        """將iTick K線JSON轉換為標準OHLCV DataFrame"""
        df = pd.DataFrame(kline_list)
        if df.empty:
            return df
        
        # 字段映射：iTick響應字段 -> 標準列名
        df.rename(columns={
            'o': 'open',
            'h': 'high', 
            'l': 'low',
            'c': 'close',
            'v': 'volume',
            't': 'timestamp'
        }, inplace=True)
        
        # 轉換時間戳並設為索引
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        df.set_index('timestamp', inplace=True)
        
        # 確保數值類型
        for col in ['open', 'high', 'low', 'close', 'volume']:
            df[col] = pd.to_numeric(df[col])
        
        return df
    
    def _generate_mock_data(self, symbol, limit):
        """生成模擬PSX數據（僅供本教程演示，實際使用時請替換為真實API響應）"""
        print(f"[模擬模式] 為 {symbol} 生成{limit}條測試數據")
        dates = pd.date_range(
            end=datetime.now(), 
            periods=limit, 
            freq='D'
        )
        np.random.seed(hash(symbol) % 100)
        price = 100 + np.cumsum(np.random.randn(limit) * 0.5)
        df = pd.DataFrame({
            'open': price * (1 + np.random.randn(limit)*0.01),
            'high': price * (1 + np.abs(np.random.randn(limit)*0.02)),
            'low': price * (1 - np.abs(np.random.randn(limit)*0.02)), 
            'close': price,
            'volume': np.random.randint(100000, 500000, limit)
        }, index=dates)
        return df

    

      import websocket
import json
import threading

class PSXRealtimeFeed:
    """PSX即時行情訂閱（WebSocket適配層）"""
    
    def __init__(self, api_key):
        self.api_key = api_key
        self.ws = None
        self.subscribers = []
    
    def connect(self):
        ws_url = "wss://api.itick.org/stock"  # 復用股票WS端點
        headers = {"token": self.api_key}
        self.ws = websocket.WebSocketApp(
            ws_url,
            header=headers,
            on_open=self._on_open,
            on_message=self._on_message,
            on_error=self._on_error
        )
        # 異步運行
        threading.Thread(target=self.ws.run_forever, daemon=True).start()
    
    def subscribe(self, symbols):
        """訂閱PSX股票即時行情"""
        if not self.ws:
            self.connect()
        
        # PSX股票按自定義格式包裝
        params = ",".join([f"{sym}$PK" for sym in symbols])
        sub_msg = {
            "ac": "subscribe",
            "params": params,
            "types": "quote,tick"  # 報價+逐筆成交
        }
        self.ws.send(json.dumps(sub_msg))
        print(f"已訂閱PSX即時行情: {symbols}")
    
    def _on_message(self, ws, message):
        data = json.loads(message)
        # 分發至已註冊的回調函數
        for cb in self.subscribers:
            cb(data)

    

      import pandas as pd
import numpy as np
import talib

class PSXDualMovingAverageStrategy:
    """PSX市場雙均線策略（適配卡拉奇交易所交易規則）"""
    
    def __init__(self, fast_period=20, slow_period=60):
        self.fast_period = fast_period
        self.slow_period = slow_period
        self.position = 0
        self.trades = []
    
    def calculate_signals(self, df):
        """生成交易信號"""
        # 計算移動平均線
        df['MA_FAST'] = talib.SMA(df['close'], self.fast_period)
        df['MA_SLOW'] = talib.SMA(df['close'], self.slow_period)
        
        # 判斷均線位置
        df['cross_above'] = (df['MA_FAST'] > df['MA_SLOW']) & \
                            (df['MA_FAST'].shift(1) <= df['MA_SLOW'].shift(1))
        df['cross_below'] = (df['MA_FAST'] < df['MA_SLOW']) & \
                            (df['MA_FAST'].shift(1) >= df['MA_SLOW'].shift(1))
        
        # 信號編碼：1=買入，-1=賣出，0=無操作
        df['signal'] = 0
        df.loc[df['cross_above'], 'signal'] = 1
        df.loc[df['cross_below'], 'signal'] = -1
        
        return df
    
    def run_backtest(self, df, initial_capital=1000000):
        """
        執行回測（巴基斯坦盧比PKR）
        PSX交易規則：最小交易單位100股，T+2結算
        """
        df = self.calculate_signals(df.copy())
        
        # 初始化投資組合
        portfolio = pd.DataFrame(index=df.index)
        portfolio['price'] = df['close']
        portfolio['signal'] = df['signal']
        portfolio['cash'] = initial_capital
        portfolio['holdings'] = 0
        portfolio['total'] = initial_capital
        
        current_position = 0
        
        for i in range(1, len(portfolio)):
            # 默認繼承前值
            portfolio.loc[portfolio.index[i], 'cash'] = portfolio.loc[portfolio.index[i-1], 'cash']
            portfolio.loc[portfolio.index[i], 'holdings'] = portfolio.loc[portfolio.index[i-1], 'holdings']
            
            # 買入信號
            if portfolio['signal'].iloc[i] == 1 and current_position == 0:
                price = portfolio['price'].iloc[i]
                max_spend = portfolio['cash'].iloc[i-1] * 0.9
                shares = int(max_spend // price // 100 * 100)  # 整手交易
                
                if shares >= 100:
                    cost = shares * price
                    portfolio.loc[portfolio.index[i], 'cash'] = portfolio['cash'].iloc[i-1] - cost
                    portfolio.loc[portfolio.index[i], 'holdings'] = shares
                    current_position = shares
                    self.trades.append(('BUY', portfolio.index[i], price, shares))
            
            # 賣出信號
            elif portfolio['signal'].iloc[i] == -1 and current_position > 0:
                price = portfolio['price'].iloc[i]
                shares = current_position
                proceeds = shares * price
                portfolio.loc[portfolio.index[i], 'cash'] = portfolio['cash'].iloc[i-1] + proceeds
                portfolio.loc[portfolio.index[i], 'holdings'] = 0
                current_position = 0
                self.trades.append(('SELL', portfolio.index[i], price, shares))
            
            # 更新總資產
            portfolio.loc[portfolio.index[i], 'total'] = \
                portfolio['cash'].iloc[i] + portfolio['holdings'].iloc[i] * portfolio['price'].iloc[i]
        
        self.portfolio = portfolio
        return portfolio

    

      # 初始化數據獲取器
fetcher = PSXDataFetcher(api_key=ITICK_API_KEY)

# 獲取OGDC歷史日線數據（模擬近3年）
df_ogdc = fetcher.get_historical_data("OGDC", ktype=7, limit=750)

# 運行策略回測
strategy = PSXDualMovingAverageStrategy(fast_period=20, slow_period=60)
portfolio = strategy.run_backtest(df_ogdc, initial_capital=1000000)

# 計算績效指標
total_return = (portfolio['total'].iloc[-1] / portfolio['total'].iloc[0] - 1) * 100
sharpe_ratio = np.sqrt(252) * (portfolio['total'].pct_change().mean() / portfolio['total'].pct_change().std())
max_drawdown = (portfolio['total'] / portfolio['total'].cummax() - 1).min() * 100

print("="*50)
print("📊 PSX雙均線策略回測報告")
print("="*50)
print(f"標的資產: OGDC (巴基斯坦油氣開發公司)")
print(f"回測區間: {df_ogdc.index[0].date()} → {df_ogdc.index[-1].date()}")
print(f"初始資本: 1,000,000 PKR")
print(f"最終權益: {portfolio['total'].iloc[-1]:,.0f} PKR")
print(f"總收益率: {total_return:.2f}%")
print(f"年化夏普: {sharpe_ratio:.2f}")
print(f"最大回撤: {max_drawdown:.2f}%")
print("="*50)

# 輸出交易記錄
print("\n📝 交易明細:")
for i, trade in enumerate(strategy.trades[:10], 1):
    action, date, price, shares = trade
    print(f"  {i}. {action} {date.date()} @ {price:.2f} PKR, {shares}股")

    

      import matplotlib.pyplot as plt

plt.figure(figsize=(14, 8))

# 子圖1：價格與均線
plt.subplot(2, 1, 1)
plt.plot(df_ogdc.index, df_ogdc['close'], label='OGDC收盤價', alpha=0.7)
ma_values = strategy.calculate_signals(df_ogdc)
plt.plot(ma_values.index, ma_values['MA_FAST'], label=f'{strategy.fast_period}日均線', linestyle='--')
plt.plot(ma_values.index, ma_values['MA_SLOW'], label=f'{strategy.slow_period}日均線', linestyle='--')

# 標註買賣點
buy_signals = [i for i in range(len(portfolio)) if portfolio['signal'].iloc[i] == 1]
sell_signals = [i for i in range(len(portfolio)) if portfolio['signal'].iloc[i] == -1]
plt.scatter(portfolio.index[buy_signals], portfolio['price'].iloc[buy_signals], 
            marker='^', color='green', s=100, label='買入信號')
plt.scatter(portfolio.index[sell_signals], portfolio['price'].iloc[sell_signals], 
            marker='v', color='red', s=100, label='賣出信號')
plt.title('OGDC 雙均線策略交易信號 (PSX)')
plt.legend()
plt.grid(alpha=0.3)

# 子圖2：權益曲線
plt.subplot(2, 1, 2)
plt.plot(portfolio.index, portfolio['total'], label='投資組合價值', color='navy')
plt.fill_between(portfolio.index, portfolio['total'], alpha=0.2)
plt.title('權益曲線 (PKR)')
plt.legend()
plt.grid(alpha=0.3)

plt.tight_layout()
plt.show()

    

      class PSXLiveTrader:
    """PSX即時交易監控器（模擬執行）"""
    
    def __init__(self, strategy, fetcher):
        self.strategy = strategy
        self.fetcher = fetcher
        self.current_position = 0
    
    def on_bar_update(self, symbol, new_bar):
        """處理新到達的K線"""
        # 獲取最新數據窗口
        df = self.fetcher.get_historical_data(symbol, ktype=5, limit=100)
        df = self.strategy.calculate_signals(df)
        
        # 檢查最新信號
        latest_signal = df['signal'].iloc[-1]
        last_price = df['close'].iloc[-1]
        
        if latest_signal == 1 and self.current_position == 0:
            print(f"🟢 買入信號: {symbol} @ {last_price:.2f} PKR")
            # 此處可接入券商API執行實盤下單
            self.current_position = 1
            
        elif latest_signal == -1 and self.current_position == 1:
            print(f"🔴 賣出信號: {symbol} @ {last_price:.2f} PKR")
            self.current_position = 0

    

      # 偽代碼示例：轉換信號為券商指令
order_payload = {
    "broker_code": "XYZ",
    "symbol": "OGDC",
    "side": "BUY",
    "order_type": "LIMIT",
    "price": 125.50,
    "quantity": 5000,
    "validity": "DAY"
}
# 通過券商提供的REST API提交

    

      # 遍歷不同均線組合
results = []
for fast in [10, 20, 30, 50]:
    for slow in [50, 60, 120, 200]:
        if fast >= slow: continue
        strat = PSXDualMovingAverageStrategy(fast, slow)
        port = strat.run_backtest(df_ogdc, initial_capital=1000000)
        ret = (port['total'].iloc[-1] / 1000000 - 1) * 100
        results.append({'fast': fast, 'slow': slow, 'return': ret})

# 輸出最佳參數組合
best = max(results, key=lambda x: x['return'])
print(f"最優參數: 快線{best['fast']}, 慢線{best['slow']}, 收益率{best['return']:.2f}%")

    

      class PSXRiskManager:
    """PSX市場風控規則"""
    
    @staticmethod
    def check_circuit_breaker(price_change_pct):
        """PSX漲跌停限制（指數成分股±5%）"""
        return abs(price_change_pct) <= 5.0
    
    @staticmethod
    def position_sizing(equity, atr, risk_per_trade=0.02):
        """基於ATR的動態倉位計算"""
        risk_amount = equity * risk_per_trade
        shares = int(risk_amount / atr / 100) * 100
        return max(shares, 100)
    
    @staticmethod
    def trading_hours_filter(dt):
        """PSX交易時段過濾（週一至週五 09:30-15:30 PST）"""
        if dt.weekday() >= 5:
            return False
        pst_hour = dt.hour + 5  # UTC+5轉換
        return (9 <= pst_hour <= 15) and not (pst_hour == 15 and dt.minute > 30)

    

      # 獲取美元/巴基斯坦盧比匯率
def get_usdpkr_rate():
    url = "https://api.itick.org/forex/quote?region=GB&code=USDPKR"
    # 實際請求代碼...
    return rate