Pakistan Stock Exchange (PSX) Data via API — Real-Time Market Data & Quantitative Trading Implementation Guide

      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 authenticates via token header
    }
    try:
        print(f"🔄 Requesting PSX ticker list...")
        response = requests.get(
            url, 
            headers=headers, 
            timeout=15
        )
        
        # Handle HTTP status codes
        if response.status_code == 200:
            data = response.json()
            if data.get("code") == 200:
                symbol_list = data.get("data", [])
                print(f"✅ Successfully retrieved {len(symbol_list)} PSX tickers")
                return symbol_list
            else:
                print(f"⚠️ API business error: {data.get('msg')} (Code: {data.get('code')})")
                return None
                
        elif response.status_code == 401:
            print("❌ Authentication failed (401): Check if API key is valid/expired or if the endpoint is covered by your plan")
            print("   - Log in to https://itick.org dashboard to verify key status")
            print("   - Confirm whether your plan includes 'basic stock data' permissions")
            return None
            
        elif response.status_code == 403:
            print("❌ Insufficient permissions (403): Your account lacks access to PSX regional data")
            print("   - PSX may currently be in restricted beta; contact sales to enable")
            return None
            
        elif response.status_code == 429:
            print("❌ Rate limit exceeded (429): Retry later or upgrade plan for higher limits")
            return None
            
        else:
            print(f"❌ Unknown error: HTTP {response.status_code}")
            return None
    except Exception as e:
        print(f"❌ Request exception: {str(e)}")
        return None


def save_to_dataframe(symbol_list: List[Dict]) -> pd.DataFrame:
    """Save ticker list as Pandas DataFrame for analysis"""
    if not symbol_list:
        return pd.DataFrame()
    
    df = pd.DataFrame(symbol_list)
    print("\n📋 Preview of PSX Ticker List:")
    print(df.head())
    print(f"\nTotal records: {len(df)}")
    return df


# ===== Usage Example =====
if __name__ == "__main__":
    # ⚠️ Important: Load key from environment variables or config files; never hardcode
    YOUR_API_KEY = os.environ.get("ITICK_API_KEY", "")
    
    if not YOUR_API_KEY:
        print("Please set environment variable ITICK_API_KEY or assign directly (testing only)")
        # YOUR_API_KEY = "your_key_here"  # Temporarily uncomment for testing
    else:
        # Step 1: Fetch PSX ticker list
        symbols = fetch_psx_symbols(YOUR_API_KEY)
        
        # Step 2: Convert to DataFrame
        if symbols:
            df = save_to_dataframe(symbols)
            
            # Optional: Save to CSV
            df.to_csv("psx_symbols.csv", index=False)
            print("\n💾 Saved to psx_symbols.csv")
            
            # Print all ticker symbols
            print("\n🔹 PSX Ticker Symbols:")
            for code in df.get("symbol", [])[:20]:  # Show top 20
                print(f"  {code}")

    

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

class PSXDataFetcher:
    """Pakistan PSX Market Data Fetcher (based on iTick API protocol adaptation)"""
    
    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):
        """
        Retrieve PSX stock historical K-line data
        :param symbol: PSX ticker symbol (e.g., 'OGDC')
        :param ktype: K-line period (1=min, 2=5min, 3=15min, 4=30min, 5=1hr, 6=2hr, 7=4hr, 8=daily, 9=weekly, 10=monthly)
        :param limit: Number of K-lines returned (max 500)
        :return: Pandas DataFrame
        """
        # Adapt PSX market via custom region parameter
        params = {
            "region": "PK",      # Market identifier
            "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 error: {data.get('msg')}")
                return None
                
        except Exception as e:
            print(f"Request failed for {symbol}: {str(e)}")
            # Mock data generation (for demo purposes only—remove in production)
            return self._generate_mock_data(symbol, limit)
    
    def _parse_kline(self, kline_list):
        """Convert iTick K-line JSON to standard OHLCV DataFrame"""
        df = pd.DataFrame(kline_list)
        if df.empty:
            return df
        
        # Field mapping: iTick response fields → standard column names
        df.rename(columns={
            'o': 'open',
            'h': 'high', 
            'l': 'low',
            'c': 'close',
            'v': 'volume',
            't': 'timestamp'
        }, inplace=True)
        
        # Convert timestamps and set as index
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        df.set_index('timestamp', inplace=True)
        
        # Ensure numeric types
        for col in ['open', 'high', 'low', 'close', 'volume']:
            df[col] = pd.to_numeric(df[col])
        
        return df
    
    def _generate_mock_data(self, symbol, limit):
        """Generate mock PSX data (demo only—replace with real API responses in production)"""
        print(f"[Mock Mode] Generating {limit} test records for {symbol}")
        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 Real-Time Market Feed (WebSocket Adapter Layer)"""
    
    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"  # Reuse stock WebSocket endpoint
        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
        )
        # Run asynchronously
        threading.Thread(target=self.ws.run_forever, daemon=True).start()
    
    def subscribe(self, symbols):
        """Subscribe to PSX real-time market data"""
        if not self.ws:
            self.connect()
        
        # Wrap PSX tickers in custom format
        params = ",".join([f"{sym}$PK" for sym in symbols])
        sub_msg = {
            "ac": "subscribe",
            "params": params,
            "types": "quote,tick"  # Quotes + tick-by-tick trades
        }
        self.ws.send(json.dumps(sub_msg))
        print(f"Subscribed to PSX real-time feed: {symbols}")
    
    def _on_message(self, ws, message):
        data = json.loads(message)
        # Dispatch to registered callback functions
        for cb in self.subscribers:
            cb(data)

    

      import pandas as pd
import numpy as np
import talib

class PSXDualMovingAverageStrategy:
    """Dual Moving Average Strategy for PSX Market (adapted to Karachi Exchange rules)"""
    
    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):
        """Generate trading signals"""
        # Calculate moving averages
        df['MA_FAST'] = talib.SMA(df['close'], self.fast_period)
        df['MA_SLOW'] = talib.SMA(df['close'], self.slow_period)
        
        # Determine crossover conditions
        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))
        
        # Encode signals: 1=Buy, -1=Sell, 0=No Action
        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):
        """
        Execute backtest (in Pakistani Rupees PKR)
        PSX trading rules: Minimum trade size 100 shares, T+2 settlement
        """
        df = self.calculate_signals(df.copy())
        
        # Initialize portfolio
        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)):
            # Default inheritance from previous value
            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']
            
            # Buy signal
            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)  # Round to full lots
                
                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))
            
            # Sell signal
            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))
            
            # Update total assets
            portfolio.loc[portfolio.index[i], 'total'] = \
                portfolio['cash'].iloc[i] + portfolio['holdings'].iloc[i] * portfolio['price'].iloc[i]
        
        self.portfolio = portfolio
        return portfolio

    

      # Initialize data fetcher
fetcher = PSXDataFetcher(api_key=ITICK_API_KEY)

# Retrieve OGDC historical daily data (simulated ~3 years)
df_ogdc = fetcher.get_historical_data("OGDC", ktype=7, limit=750)

# Run strategy backtest
strategy = PSXDualMovingAverageStrategy(fast_period=20, slow_period=60)
portfolio = strategy.run_backtest(df_ogdc, initial_capital=1000000)

# Compute performance metrics
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 Dual Moving Average Strategy Backtest Report")
print("="*50)
print(f"Asset: OGDC (Oil & Gas Development Company Ltd.)")
print(f"Backtest Period: {df_ogdc.index[0].date()} → {df_ogdc.index[-1].date()}")
print(f"Initial Capital: 1,000,000 PKR")
print(f"Final Equity: {portfolio['total'].iloc[-1]:,.0f} PKR")
print(f"Total Return: {total_return:.2f}%")
print(f"Annualized Sharpe Ratio: {sharpe_ratio:.2f}")
print(f"Max Drawdown: {max_drawdown:.2f}%")
print("="*50)

# Output trade log
print("\n📝 Trade Details:")
for i, trade in enumerate(strategy.trades[:10], 1):
    action, date, price, shares = trade
    print(f"  {i}. {action} {date.date()} @ {price:.2f} PKR, {shares} shares")

    

      import matplotlib.pyplot as plt

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

# Subplot 1: Price and Moving Averages
plt.subplot(2, 1, 1)
plt.plot(df_ogdc.index, df_ogdc['close'], label='OGDC Closing Price', alpha=0.7)
ma_values = strategy.calculate_signals(df_ogdc)
plt.plot(ma_values.index, ma_values['MA_FAST'], label=f'{strategy.fast_period}-Day MA', linestyle='--')
plt.plot(ma_values.index, ma_values['MA_SLOW'], label=f'{strategy.slow_period}-Day MA', linestyle='--')

# Mark buy/sell signals
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='Buy Signal')
plt.scatter(portfolio.index[sell_signals], portfolio['price'].iloc[sell_signals], 
            marker='v', color='red', s=100, label='Sell Signal')
plt.title('OGDC Dual Moving Average Trading Signals (PSX)')
plt.legend()
plt.grid(alpha=0.3)

# Subplot 2: Equity Curve
plt.subplot(2, 1, 2)
plt.plot(portfolio.index, portfolio['total'], label='Portfolio Value', color='navy')
plt.fill_between(portfolio.index, portfolio['total'], alpha=0.2)
plt.title('Equity Curve (PKR)')
plt.legend()
plt.grid(alpha=0.3)

plt.tight_layout()
plt.show()

    

      class PSXLiveTrader:
    """PSX Real-Time Trading Monitor (Simulated Execution)"""
    
    def __init__(self, strategy, fetcher):
        self.strategy = strategy
        self.fetcher = fetcher
        self.current_position = 0
    
    def on_bar_update(self, symbol, new_bar):
        """Process newly arrived K-line"""
        # Retrieve latest data window
        df = self.fetcher.get_historical_data(symbol, ktype=5, limit=100)
        df = self.strategy.calculate_signals(df)
        
        # Check latest signal
        latest_signal = df['signal'].iloc[-1]
        last_price = df['close'].iloc[-1]
        
        if latest_signal == 1 and self.current_position == 0:
            print(f"🟢 Buy Signal: {symbol} @ {last_price:.2f} PKR")
            # Insert broker API execution here
            self.current_position = 1
            
        elif latest_signal == -1 and self.current_position == 1:
            print(f"🔴 Sell Signal: {symbol} @ {last_price:.2f} PKR")
            self.current_position = 0

    

      # Pseudocode example: Convert signals to broker instructions
order_payload = {
    "broker_code": "XYZ",
    "symbol": "OGDC",
    "side": "BUY",
    "order_type": "LIMIT",
    "price": 125.50,
    "quantity": 5000,
    "validity": "DAY"
}
# Submit via broker-provided REST API

    

      # Iterate over different MA combinations
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})

# Output optimal parameter combination
best = max(results, key=lambda x: x['return'])
print(f"Optimal Parameters: Fast MA {best['fast']}, Slow MA {best['slow']}, Return {best['return']:.2f}%")

    

      class PSXRiskManager:
    """PSX Market Risk Rules"""
    
    @staticmethod
    def check_circuit_breaker(price_change_pct):
        """PSX circuit breaker limit (±5% for index constituents)"""
        return abs(price_change_pct) <= 5.0
    
    @staticmethod
    def position_sizing(equity, atr, risk_per_trade=0.02):
        """Dynamic position sizing based on ATR"""
        risk_amount = equity * risk_per_trade
        shares = int(risk_amount / atr / 100) * 100
        return max(shares, 100)
    
    @staticmethod
    def trading_hours_filter(dt):
        """Filter for PSX trading hours (Mon-Fri 09:30–15:30 PST)"""
        if dt.weekday() >= 5:
            return False
        pst_hour = dt.hour + 5  # UTC+5 conversion
        return (9 <= pst_hour <= 15) and not (pst_hour == 15 and dt.minute > 30)

    

      # Retrieve USD/PKR exchange rate
def get_usdpkr_rate():
    url = "https://api.itick.org/forex/quote?region=GB&code=USDPKR"
    # Actual request code...
    return rate