馬來西亞股票 API 接入指南：即時行情與歷史數據

      import requests

API_TOKEN = "your_token_here"  # 請替換為您的實際 Token
BASE_URL = "https://api.itick.org"

def get_malaysia_quote(symbol):
    url = f"{BASE_URL}/stock/quote"
    params = {"region": "MY", "code": symbol}
    headers = {"token": API_TOKEN}

    try:
        resp = requests.get(url, params=params, headers=headers, timeout=5)
        data = resp.json()
        if data.get("code") == 0:
            quote = data["data"]
            print(f"📈 {quote.get('n')} ({quote.get('s')})")
            print(f"最新價: {quote.get('ld')} MYR")
            print(f"漲跌幅: {quote.get('chp')}%")
            print(f"成交量: {quote.get('v')}")
            return quote
        else:
            print(f"API 錯誤: {data.get('msg')}")
    except Exception as e:
        print(f"請求異常: {e}")

# 呼叫範例
get_malaysia_quote("MAYBANK")  # Maybank

    

      def get_malaysia_kline(symbol, ktype=8, limit=100):
    """ktype：8=日線, 9=週線, 10=月線, 1~5=分鐘線"""
    url = f"{BASE_URL}/stock/kline"
    params = {"region": "MY", "code": symbol, "kType": ktype, "limit": limit}
    headers = {"token": API_TOKEN}

    resp = requests.get(url, params=params, headers=headers)
    data = resp.json()
    if data.get("code") == 0:
        kline_list = data.get("data", [])
        print(f"取得 {len(kline_list)} 條 K 線數據")
        for item in kline_list[-5:]:  # 顯示最近 5 條
            print(f"時間:{item['t']} 開:{item['o']} 高:{item['h']} 低:{item['l']} 收:{item['c']} 量:{item['v']}")
    else:
        print(f"錯誤: {data.get('msg')}")

get_malaysia_kline("MAYBANK", limit=10)  # 最近 10 個交易日

    

      import websocket
import json
import threading
import time

WS_URL = "wss://api.itick.org/stock"
API_TOKEN = "your_token_here"

def on_message(ws, message):
    data = json.loads(message)
    if "data" in data:
        md = data["data"]
        if md.get("type") == "quote":
            print(f"[{md['s']}] 最新: {md['ld']} MYR | 漲跌:{md['chp']}%")

def on_open(ws):
    # 訂閱格式：股票代碼$MY
    subscribe = {
        "ac": "subscribe",
        "params": "MAYBANK$MY,PBBANK$MY,TENAGA$MY",
        "types": "quote"
    }
    ws.send(json.dumps(subscribe))

def send_ping(ws):
    while True:
        time.sleep(30)
        ws.send(json.dumps({"ac": "ping", "params": str(int(time.time()*1000))}))

ws = websocket.WebSocketApp(WS_URL, header={"token": API_TOKEN},
                            on_open=on_open, on_message=on_message)
threading.Thread(target=send_ping, args=(ws,), daemon=True).start()
ws.run_forever()