2026 澳洲股市 ASX 實時行情API接入指南

      import requests

API_TOKEN = 'your_api_token_here'
BASE_URL = 'https://api.itick.org'

def get_real_time_quote(region, code):
    headers = {
        'accept': 'application/json',
        'token': API_TOKEN
    }
    url = f'{BASE_URL}/stock/quote?region={region}&code={code}'
    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        data = response.json()['data']  # 響應數據為 JSON 對象  字段可參考官方文檔
        return data
    else:
        print(f"Error: {response.status_code}")
        return None

# 示例：獲取 BHP.AX 的實時行情（region=AU, code=BHP）
quote = get_real_time_quote('AU', 'BHP')
if quote:
    print(f"Symbol: {quote['s']}")
    print(f"Current Price: {quote['ld']}")
    print(f"Volume: {quote['v']}")
    print(f"Change: {quote['ch']} ({quote['chp']}%)")

    

      import websocket
import json
import threading
import time

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

def on_message(ws, message):
    data = json.loads(message)
    if data.get("code") == 1 and data.get("msg") == "Connected Successfully":
        print("Connected successfully")
    elif data.get("resAc") == "auth" and data.get("code") == 1:
        print("Authentication successful")
        subscribe(ws)
    elif data.get("resAc") == "subscribe" and data.get("code") == 1:
        print("Subscription successful")
    elif data.get("data"):
        market_data = data["data"]  # 響應數據為 JSON 對象，字段可參考官方文檔
        data_type = market_data.get("type")
        symbol = market_data.get("s")
        print(f"{data_type.upper()} data for {symbol}:", market_data)

def on_error(ws, error):
    print("Error:", error)

def on_close(ws, close_status_code, close_msg):
    print("Connection closed")

def on_open(ws):
    print("WebSocket connection opened")

def subscribe(ws):
    subscribe_msg = {
        "ac": "subscribe",
        "params": "BHP$AU",
        "types": "tick,quote,depth"
    }
    ws.send(json.dumps(subscribe_msg))
    print("Subscribe message sent")

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

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

    

      import requests
import pandas as pd

API_TOKEN = 'your_api_token_here'
BASE_URL = 'https://api.itick.org'

def get_historical_data(region, code, kType='8', limit=10, et=None):
    headers = {
        'accept': 'application/json',
        'token': API_TOKEN
    }
    url = f'{BASE_URL}/stock/kline?region={region}&code={code}&kType={kType}&limit={limit}'
    if et:
        url += f'&et={et}'
    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        data = response.json()['data']  # 響應數據為數組 
        df = pd.DataFrame(data, columns=['t', 'o', 'h', 'l', 'c', 'v', 'tu'])
        return df
    else:
        print(f"Error: {response.status_code}")
        return None

# 示例：獲取 CSL.AX 的日線歷史數據（region=AU, code=CSL, kType=8 為日K）
historical = get_historical_data('AU', 'CSL')
if historical is not None:
    print(historical.head())  # 顯示前幾行數據