马来西亚股票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, json, threading, 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()