港股和美股免費api接口(整理自網絡)

 

1.港股

實時行情

import requests
hk_sina_stock_list_url = "http://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/Market_Center.getHKStockData"
hk_sina_stock_dict_payload = {
    "page": "1",
    "num": "3000",
    "sort": "symbol",
    "asc": "1",
    "node": "qbgg_hk",
    "_s_r_a": "page"
}
def sina_hk_real():
    """
    新浪財經-港股的所有港股的實時行情數據
    http://vip.stock.finance.sina.com.cn/mkt/#qbgg_hk
    :return: 實時行情數據
    :rtype: []
    """
    res = requests.get(hk_sina_stock_list_url, params=hk_sina_stock_dict_payload)
    if res.status_code != 200:
        logging.error(f"sina_hk_real/status_code:{res.status_code}/text:{res.text}")
        return []
    else:
        try:
            data_json = res.json()
            """
            {'symbol': '00021',  # 港股代碼
            'name': '大中華地產控股',  # 中文名稱
            'engname': 'GREAT CHI PPT',  # 英文名稱
            'tradetype': 'EQTY',  # 交易類型
            'lasttrade': '0.000',  # 最新價
            'prevclose': '0.118',  # 前一個交易日收盤價
            'open': '0.000',  # 開盤價
            'high': '0.000',  # 最高價
            'low': '0.000',  # 最低價
            'volume': '0',  # 成交量(萬)
            'currentvolume': '0',  # 每手股數
            'amount': '0',  # 成交額(萬)
            'ticktime': '2022-04-08 10:54:17',  # 當前數據時間戳
            'buy': '0.115',  # 買一
            'sell': '0.120',  # 賣一
            'high_52week': '0.247',  # 52周最高價
            'low_52week': '0.110',  # 52周最低價
            'eps': '-0.003',   # 每股收益
            'dividend': None,   # 股息
            'stocks_sum': '3975233406', 
            'pricechange': '0.000',  # 漲跌額
            'changepercent': '0.0000000',  # 漲跌幅
            'market_value': '0.000',  # 港股市值
            'pe_ratio': '0.0000000'
            }
            """
        except Exception as e:
            logging.error(f"sina_hk_real/error: res.json()/detail:{e.__str__()}")
            return []
        res = [
            {"stock_code": d["symbol"],
             "name": d["name"],
             "eng_name": d["engname"],
             "date": d["ticktime"][:10],
             "time": d["ticktime"][11:],
             "now": float(d["lasttrade"]),
             "open": float(d["open"]),
             "close": float(d["prevclose"]),
             "high": float(d["high"]),
             "low": float(d["low"]),
             "volume": float(d["amount"]) * 10000,
             "turnover": float(d["volume"]) * 10000,
             "buy": float(d["buy"]),
             "sell": float(d["sell"]),
             "change": float(d["pricechange"]),
             "change_rate": float(d["changepercent"]),
             "stock_type": "hk"}
            for d in data_json
        ]
    return res

股票代碼

def get_hk_stock_codes():
    stocks = sina_hk_real()
    return [{"stock_code": i["stock_code"],
             "name": i["name"],
             "eng_name": i["eng_name"],
             "stock_type": "hk"}
            for i in stocks]

1分鍾級k線

import requests
hk_tencent_minute = "https://web.ifzq.gtimg.cn/appstock/app/minute/query?code={}"
def tencent_hk_minute(stock_code, latest_minute=True):
    prefix_code = "hk" + stock_code
    res = requests.get(hk_tencent_minute.format(prefix_code), headers=headers)
    if res.status_code != 200:
        logging.error(f"tencent_hk_minute/status_code:{res.status_code}/text:{res.text}")
        return []
    else:
        try:
            data_json = res.json()
        except Exception as e:
            logging.error(f"tencent_hk_minute/error: res.json()/detail:{e.__str__()}")
            return []
        if "data" in data_json \
                and prefix_code in data_json["data"] \
                and "data" in data_json["data"][prefix_code] \
                and "data" in data_json["data"][prefix_code]["data"] \
                and data_json["data"][prefix_code]["data"]["data"]:
            date = data_json["data"][prefix_code]["data"]["date"]
            date = date[:4] + "-" + date[4:6] + "-" + date[6:]
            res = []
            prev = 0
            for d in data_json["data"][prefix_code]["data"]["data"]:
                d = d.split(" ")
                res.append({
                    "datetime": date + " " + d[0][:2] + ":" + d[0][2:] + ":00",
                    "close": float(d[1]),
                    "volume": float(d[2]) - prev,
                    "stock_type": "hk"
                })
                prev = float(d[2]) - prev

            if latest_minute:
                return res[-1]
            else:
                return res
        else:
            return []
print(tencent_hk_minute("00001"))

天級別k線

import requests
from py_mini_racer import py_mini_racer
hk_sina_stock_hist_url = "https://finance.sina.com.cn/stock/hkstock/{}/klc_kl.js"

def sina_hk_daily(symbol, adjust=""):
    """
    新浪財經-港股-個股的歷史行情數據
    https://stock.finance.sina.com.cn/hkstock/quotes/02912.html
    """
    res = requests.get(hk_sina_stock_hist_url.format(symbol))
    js_code = py_mini_racer.MiniRacer()
    js_code.eval(hk_js_decode)
    dict_list: list = js_code.call(
        "d", res.text.split("=")[1].split(";")[0].replace('"', "")
    )  # 執行js解密代碼
    dict_list.sort(key=lambda x: x["date"])
    
    if adjust == "qfq":
        # 使用前復權
        res = requests.get(hk_sina_stock_hist_qfq_url.format(symbol))
        try:
            qfq = json.loads(res.text.split("=")[1].split("\n")[0])["data"]
            if len(qfq) == 1:
                return dict_list
        except SyntaxError as e:
            return dict_list
        qfq = sorted(map(lambda x: (x["d"], float(x["f"])), qfq))
        n = len(qfq)
        qfq_index = 0
        for data_index in range(len(dict_list)):
            while qfq_index + 1 < n and qfq[qfq_index + 1][0] <= dict_list[data_index]["date"][:10]:
                qfq_index += 1
            dict_list[data_index]["high"] *= qfq[qfq_index][1]
            dict_list[data_index]["low"] *= qfq[qfq_index][1]
            dict_list[data_index]["open"] *= qfq[qfq_index][1]
            dict_list[data_index]["close"] *= qfq[qfq_index][1]
    return dict_list
 print(sina_hk_daily("00001"))
def tencent_hk_daily(symbol, day=0):
    prefix_code = "hk" + symbol
    res = requests.get(hk_tencent_daily.format(prefix_code, day), headers=headers)
    if res.status_code != 200:
        logging.error(f"tencent_hk_daily/status_code:{res.status_code}/text:{res.text}")
        return []
    else:
        try:
            data_json = json.loads(res.text.split("=")[1].split("\n")[0])
        except Exception as e:
            logging.error(f"tencent_hk_daily/error: res.json()/detail:{e.__str__()}")
            return []
        if "data" in data_json \
                and prefix_code in data_json["data"] \
                and "day" in data_json["data"][prefix_code]:
            res = []
            for d in data_json["data"][prefix_code]["day"]:
                res.append({
                        "day": d[0],
                        "open": float(d[1]),
                        "close": float(d[2]),
                        "high": float(d[3]),
                        "low": float(d[4]),
                        "volume": float(d[5])
                    })
            return res
        else:
            return []

2.美股

實時行情

import requests
import pandas as pd
def em_us_real():
    """
    東方財富-美股-實時行情
    http://quote.eastmoney.com/center/gridlist.html#us_stocks
    :return: 美股-實時行情; 延遲 15 min
    "f1": "_",
    "f2": "最新價",
    "f3": "漲跌幅",
    "f4": "漲跌額",
    "f5": "成交量",
    "f6": "成交額",
    "f7": "振幅",
    "f8": "換手率",
    "f9": "-",
    "f10": "_",
    "f11": "_",
    "f12": "簡稱",
    "f13": "編碼",
    "f14": "名稱",
    "f15": "最高價",
    "f16": "最低價",
    "f17": "開盤價",
    "f18": "昨收價",
    "f20": "總市值",
    "f21": "_",
    "f22": "_",
    "f23": "_",
    "f24": "_",
    "f25": "_",
    "f26": "_",
    "f33": "_",
    "f62": "_",
    "f115": "市盈率",
    "f124": "時間戳",
    "f128": "_",
    "f140": "_",
    "f141": "_",
    "f136": "_",
    "f152": "_",
    """
    url = "http://72.push2.eastmoney.com/api/qt/clist/get"
    params = {
        "pn": "1",
        "pz": "20000",
        "po": "1",
        "np": "1",
        "ut": "bd1d9ddb04089700cf9c27f6f7426281",
        "fltt": "2",
        "invt": "2",
        "fid": "f3",
        "fs": "m:105,m:106,m:107",
        "fields": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f12,f13,f14,f15,f16,f17,f18,f20,f21,"
                  "f23,f24,f25,f26,f22,f33,f11,f62,f128,f136,f115,f124,f152",
        "_": str(int(time.time() * 1000)),
    }
    res = requests.get(url, params=params)
    if res.status_code != 200:
        logging.error(f"em_us_real/status_code:{res.status_code}/text:{res.text}")
        return []
    else:
        try:
            data_json = res.json()
        except Exception as e:
            logging.error(f"em_us_real/error: res.json()/detail:{e.__str__()}")
            return []
        if "data" in data_json and "diff" in data_json["data"] and data_json["data"]["diff"]:
            res = []
            for d in data_json["data"]["diff"]:
                _datetime = mystriptime(d["f124"])
                res.append({"stock_code": d["f12"],
                            "name": d["f14"],
                            "eng_name": d["f12"],
                            "date": _datetime.strftime("%Y-%m-%d"),
                            "time": _datetime.strftime("%H:%M:%M"),
                            "now": pd.to_numeric(d["f2"], errors="coerce"),
                            "open": pd.to_numeric(d["f17"], errors="coerce"),
                            "close": pd.to_numeric(d["f18"], errors="coerce"),
                            "high": pd.to_numeric(d["f15"], errors="coerce"),
                            "low": pd.to_numeric(d["f16"], errors="coerce"),
                            "volume": pd.to_numeric(d["f6"], errors="coerce"),
                            "turnover": pd.to_numeric(d["f5"], errors="coerce"),
                            "buy": pd.to_numeric(d["f2"], errors="coerce"),
                            "sell": pd.to_numeric(d["f2"], errors="coerce"),
                            "change": pd.to_numeric(d["f4"], errors="coerce"),
                            "change_rate": pd.to_numeric(d["f3"], errors="coerce"),
                            "stock_type": "us"})
            return res
        else:
            return []

美股代碼

 

def get_us_stock_codes():
    stocks = em_us_real()
    return [{"stock_code": i["stock_code"],
             "name": i["name"],
             "eng_name": i["eng_name"],
             "stock_type": "us"}
            for i in stocks]

1分鍾級k線

import requests
headers = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/100.0.4896.75 "
        "Safari/537.36"
    )
}
def tencent_us_minute(stock_code, latest_minute=True):
    prefix_code = "us" + stock_code + ".OQ"
    res = requests.get(us_tencent_minute.format(prefix_code), headers=headers)
    if res.status_code != 200:
        logging.error(f"tencent_hk_minute/status_code:{res.status_code}/text:{res.text}")
        return []
    else:
        try:
            data_json = res.json()
        except Exception as e:
            logging.error(f"tencent_hk_minute/error: res.json()/detail:{e.__str__()}")
            return []
        if "data" in data_json \
                and prefix_code in data_json["data"] \
                and "data" in data_json["data"][prefix_code] \
                and "data" in data_json["data"][prefix_code]["data"] \
                and data_json["data"][prefix_code]["data"]["data"]:
            date = data_json["data"][prefix_code]["data"]["date"]
            date = date[:4] + "-" + date[4:6] + "-" + date[6:]
            res = []
            prev = 0
            for d in data_json["data"][prefix_code]["data"]["data"]:
                d = d.split(" ")
                res.append({
                    "datetime": date + " " + d[0][:2] + ":" + d[0][2:] + ":00",
                    "close": float(d[1]),
                    "volume": float(d[2]) - prev,
                    "stock_type": "hk"
                })
                prev = float(d[2]) - prev

            if latest_minute:
                return res[-1]
            else:
                return res
        else:
            return []

 天k線

 

import requests
from py_mini_racer import py_mini_racer
us_sina_stock_hist_qfq_url = "https://finance.sina.com.cn/us_stock/company/reinstatement/{}_qfq.js"
us_tencent_daily = "http://web.ifzq.gtimg.cn/appstock/app/fqkline/get?_var=kline_dayqfq&param={},day,,,{},qfq"
def sina_us_daily(symbol, adjust=""):
    url = f"https://finance.sina.com.cn/staticdata/us/{symbol}"
    res = requests.get(url)
    js_code = py_mini_racer.MiniRacer()
    js_code.eval(zh_js_decode)
    dict_list = js_code.call(
        "d", res.text.split("=")[1].split(";")[0].replace('"', "")
    )  # 執行js解密代碼
    if adjust == "qfq":
        # 使用前復權
        res = requests.get(us_sina_stock_hist_qfq_url.format(symbol))
        try:
            qfq = json.loads(res.text.split("=")[1].split("\n")[0])["data"]
            if len(qfq) == 1:
                return dict_list
        except SyntaxError as e:
            return dict_list
        qfq = sorted(map(lambda x: (x["d"], float(x["f"]), float(x["c"])), qfq))
        n = len(qfq)
        qfq_index = 0
        for data_index in range(len(dict_list)):
            while qfq_index + 1 < n and qfq[qfq_index + 1][0] <= dict_list[data_index]["date"][:10]:
                qfq_index += 1
            dict_list[data_index]["high"] *= qfq[qfq_index][1]
            dict_list[data_index]["high"] += qfq[qfq_index][2]
            dict_list[data_index]["low"] *= qfq[qfq_index][1]
            dict_list[data_index]["low"] += qfq[qfq_index][2]
            dict_list[data_index]["open"] *= qfq[qfq_index][1]
            dict_list[data_index]["open"] += qfq[qfq_index][2]
            dict_list[data_index]["close"] *= qfq[qfq_index][1]
            dict_list[data_index]["close"] += qfq[qfq_index][2]
    return dict_list


def tencent_us_daily(symbol, day=0):
    prefix_code = "us" + symbol + ".OQ"
    res = requests.get(us_tencent_daily.format(prefix_code, day), headers=headers)
    if res.status_code != 200:
        logging.error(f"tencent_hk_daily/status_code:{res.status_code}/text:{res.text}")
        return []
    else:
        try:
            data_json = json.loads(res.text.split("=")[1].split("\n")[0])
        except Exception as e:
            logging.error(f"tencent_hk_daily/error: res.json()/detail:{e.__str__()}")
            return []
        if "data" in data_json \
                and prefix_code in data_json["data"] \
                and "day" in data_json["data"][prefix_code]:
            res = []
            for d in data_json["data"][prefix_code]["day"]:
                res.append({
                    "day": d[0],
                    "open": float(d[1]),
                    "close": float(d[2]),
                    "high": float(d[3]),
                    "low": float(d[4]),
                    "volume": float(d[5])
                })
            return res
        else:
            return []

 

更便捷的方法，建議使用akshare，雖然獲取速度很慢，但接口獲取方式是可以參考的。