巴基斯坦证券交易所（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, 
            params=params, 
            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