企业官网建设流程全解析

1. 从零搭一个实时港股行情小工具：到底难在哪

先说结论：用 Python 获取实时港股行情这件事，本质上不是一个“会不会写代码”的问题，而是一个“知不知道数据从哪来、怎么合规地拿、拿完怎么解析”的问题。只要把这三件事想清楚，半小时就能跑通第一版。

这几年港股市场波动大，美团、腾讯这些票经常一天内走完别人一周的行情，手动刷新行情软件根本跟不上节奏。做程序化盯盘、做异动提醒、做盘中回测，都需要先解决“实时数据从哪拿”这个源头问题。我自己最开始是写爬虫去抓网页版行情，后来发现网页接口变动太频繁，动不动就改参数和加密逻辑，维护成本极高，于是把重心转向了更稳定的数据接口方案。

这篇文章我会从数据源选型开始讲起，然后给出一套我已经跑通了很久的代码方案，最后再把我踩过的坑和排查方法全部列出来。不管你是想搞量化策略、写自动盯盘脚本，还是只是想在命令行里快速看一手行情，这篇文章都适用。

注意：本文讲的是“实时行情”，不是历史K线。实时行情的含义是盘中可以拿到接近实时的最新价、买卖盘口、成交量和成交额，延迟通常在秒级甚至毫秒级。历史K线是另外一套数据体系和接口，别搞混了。

2. 数据源选型：为什么我最后锁定了行情接口库

2.1 港股行情有哪几条路可以走

我从头到尾梳理过获取港股行情的所有可行路径，大致可以分成四类：

第一类：官方交易所数据。港交所和各类行情供应商（比如Bloomberg、Reuters）提供的官方数据最权威，延迟最低，但费用也最感人。个人开发者基本不用考虑，除非你有一个愿意掏钱的老板。而且港交所的实时行情授权机制非常严格，个人拿不到。

第二类：券商开放API。像富途、老虎、盈透这些互联网券商都开放了自己的交易接口。这类接口的好处是数据质量高、实时性强，而且还能顺便做交易。坏处是大多数都需要你在这个券商开户入金，部分接口还有资产门槛，对有代码能力但没港美股账户的同学不友好。

第三类：第三方金融数据接口库。这基本是国内开发者最常用的一条路，代表是 akshare、tushare、efinance 这类开源项目。它们做的事情简单说就是：把各类公开数据源（包括一些财经网站的接口）封装成一套统一的 Python 接口，用户不用关心底层数据从哪来，直接调用函数就能拿到结构化数据。这也是我最后选定的方案。

第四类：自己逆向爬虫。抓网页版行情接口然后自己解析。这条路我强烈不建议新手碰。原因有几点：接口经常变，今天能用的接口明天就返回密文；这类接口普遍有风控和反爬策略，IP访问频繁了容易被封；数据解析逻辑需要自己维护，一套接口失效就要重新逆向，时间成本远大于收益。除非你是公司里有专门的反爬团队，否则个人做这个就是拿时间换数据，非常不划算。

2.2 我为什么选 akshare 而不是自己写爬虫

akshare 是我用下来综合成本最低的方案。核心原因有三条：

第一，它封装好了大部分数据清洗逻辑。你传入股票代码，它返回给你的就是干净的 DataFrame，列名、索引、数据格式都是统一的，不需要自己处理 JSON 里的嵌套结构。对于行情这种强时间敏感的数据，这种“拿来即用”的体验非常重要。

第二，社区的迭代速度快。港股、A股、美股的数据接口经常因为上游数据源变更而出问题，akshare 的维护者一般会在几天内修复。你自己写爬虫的话，从发现接口挂了到逆向出新接口，至少得折腾一晚上。

第三，它对新手非常友好。代码风格统一，文档里每个数据接口都有对应的说明和示例。遇到问题去 GitHub Issues 搜一下，基本都能找到解决方案。

当然，akshare 也不是万能的，它也有自身的局限。比如有些接口的数据延迟会比其他渠道高一些，极端行情下偶尔会抽风。但考虑到它的获取成本是零，这个妥协是可以接受的。而且我下面给出的方案设计，已经把这种风险降到很低的程度了。

2.3 选型之前需要搞清楚的几个问题

在你开始动手以前，有几个问题需要先想明白，这会直接影响你的接口选型和代码设计：

你的应用场景是“准实时”还是“分钟级”？如果是做短线交易决策，可能需要秒级数据，那免费接口可能不够用；如果只是做一个盘中每隔几分钟刷新一次的观察工具，那免费接口完全够用。

你的历史回测需要什么粒度？如果回测需要分钟级甚至更高频的数据，建议不要指望实时行情接口，你要找的是专门的历史数据接口，两者的领域完全不同。

你对数据源稳定性的容忍度有多高？如果这个脚本是放在线上环境天天跑的数据管道，建议做多数据源冗余；如果只是自己本地盯盘用，一个主数据源就够了。

想清楚这几个问题，后面写代码就不会反反复复改方案。我见过太多人，代码写到一半发现接口不满足需求，又回头换数据源，白白浪费一下午。

3. 核心前置准备：环境、依赖与港股代码的坑

3.1 环境准备与依赖安装

我默认你已经装好了 Python 3.8 以上的环境，装好的可以直接跳过这一段，没装好的可以按照下面的流程操作。

我用的是 akshare 加上 pandas 的组合。akshare 负责数据获取，pandas 负责数据处理。安装命令如下：

pip install akshare pandas

如果国内网络下载慢，可以使用国内镜像源，实测速度快很多：

pip install akshare pandas -i https://pypi.tuna.tsinghua.edu.cn/simple

这里有个版本注意事项：akshare 迭代非常快，很多接口的行为会对版本敏感。建议安装完成后确认一下版本号：

import akshare as ak print(ak.__version__)

如果代码运行报错说找不到某个函数，大概率是 akshare 最近更新时改了函数名。去 GitHub 仓库查一下最新版文档，问题一般都能解决。

另外，我强烈建议用虚拟环境来管理依赖，避免不同项目的包版本互相污染。用 venv 创建和激活虚拟环境的命令如下：

python -m venv hk_quote_env # Windows hk_quote_env\Scripts\activate # macOS / Linux source hk_quote_env/bin/activate

3.2 港股代码规则：这是新手最容易踩的第一个坑

A股股民的惯性思维是六位数字代码，比如平安银行 000001，贵州茅台 600519。但港股的代码规则完全不一样，最大的区别有两点：

第一，港股代码是五位数，比如腾讯控股是 00700，阿里巴巴是 09988，美团是 03690。

第二，历史原因导致有些股票代码是四位数的旧式代码，比如汇丰控股是 00005，长和是 00001。这类代码在 akshare 中也需要补齐为五位：00001 和 00005 这种，补齐以后就是 00001 和 00005。

这里有个最容易搞错的地方：港股代码转换成 akshare 能用的格式，一般需要在代码前面加一个大写字母“h”前缀，比如“h00700”。不是加“hk”，也不是加“HK”，就是小写的“h”。这个细节如果记不住，很影响获取数据的效率。

为什么 akshare 要这样设计？主要原因是它对 A 股、港股和美股用的是同一套接口逻辑，用一个前缀来区分市场，内部才能知道该去哪个数据源拉数据。你在传代码的时候如果不加前缀，接口很可能拿到一个 A 股代码去查，查不到数据就抛异常。

下面我把常见的一个对照表列出来，方便你理解：

注意：有些港股代码是纯数字，前面没有去掉逗号的数字是很正常的。获取数据以前，建议先打印一下转换结果，确认代码格式无误再发起请求，能省掉很多调试时间。

3.3 交易时间与特殊规则：行情数据获取的前提知识

港股交易时间与 A 股不完全一样，这直接决定了你什么时候该拉数据、拉到的数据有没有意义。

港股连续交易时间是上午 9:30 至中午 12:00，下午 13:00 至 16:00。开盘前有竞价时段，收盘后也有收市竞价时段。如果你是拿这些数据进行盘中监控，那么在 12:00 到 13:00 之间拉到的数据就是午间休市的状态，数据不会更新，如果你拿这段时间的数据去做什么计算，就可能得出错误的结论。

另外，港股不设涨跌幅限制，极端情况下一天涨跌百分之几十都很正常。但这不是说不用管风险，恰恰相反，因为没有涨跌停的“缓冲”，实时行情的及时性对港股交易更关键。也正因为如此，做自动化监控时对数据源的实时性要求会更高。

港股还存在“最小变动单位”的概念，不同价位区间的股票，报价的最小跳动单位不一样。比如股价低于 0.25 港元的股票，最小变动单位是 0.001；股价在 10 到 20 港元之间的股票，最小变动单位是 0.02。这个常识在后续处理买卖盘口数据时也许用得上，先记在这里。

4. 完整实操：写一个能用的实时行情采集脚本

4.1 第一步：先用现成接口拉一只股票的实时行情

我先展示一段最精简的代码，先把一条“能跑通”的链路跑通，后面再逐步优化。

import akshare as ak # 获取腾讯控股的实时行情快照 df = ak.stock_hk_spot_em() print(df.head())

这段代码不传任何股票代码，它会一次性返回所有港股的全市场快照。放到 akshare 的内部逻辑里，它会请求一个包含港股全市场的实时行情数据源，包括最新价、涨跌幅、成交量、成交额、最高最低价、换手率等信息。

如果你只是想快速看看某个股票，不用在代码层面做太多，直接在打印结果里过滤自己关注的代码就行：

target_df = df[df["代码"] == "00700"] print(target_df)

这里要注意的是，接口返回的“代码”列是原始的五位数字字符串，不带“h”前缀。所以用这个接口过滤时直接用原始代码即可，不用加“h”。

4.2 第二步：单票多次请求的正确姿势（别把接口当数据库轮询）

全市场快照这个接口，胜在简单，但有两个问题：第一是它拉全市场数据体积比较大，网络耗时可能达到几秒；第二是如果你只关心几只票，每次都拉全市场，不仅浪费带宽，而且高频调用容易触发上游数据源的风控。

所以，当你的监控标的数量在十只以内时，更合理的做法是写一个循环，依次请求每只票的快照。akshare 提供了一只港股实时行情快照的接口，函数名是stock_hk_spot_em还是stock_hk_spot按照你实际版本为准，各版本略有差异。我这边使用的是按代码取快照的方式，下面是结合常见接口的通用写法：

import akshare as ak import pandas as pd def get_hk_quote_list(symbols): all_data = [] for s in symbols: try: df = ak.stock_hk_hist_min_em(symbol=s, period="1") # 这里只取最后一行的最新成交信息作为示意 if df is not None and len(df) > 0: latest = df.iloc[-1] all_data.append({ "symbol": s, "time": latest["时间"], "price": latest["最新价"], }) except Exception as e: print(f"{s} 获取失败: {e}") continue return pd.DataFrame(all_data) my_symbols = ["h00700", "h09988", "h03690", "h01810"] result = get_hk_quote_list(my_symbols) print(result)

这个方案有个好处：按票隔离异常。一只票的接口挂了，不会影响其他票的数据获取，每次请求也能控制在上游数据源的合理访问频次内。

提示：上述接口stock_hk_hist_min_em返回的是分钟历史数据，并不是毫秒级的实时行情推送。如果你需要“实时”两个字，有两种理解：一种是“准实时拉取”，就是你每隔一段时间请求一次，拿到的数据基本上是当前最新一分钟的数据；另一种是交易所级别的实时推送，需要接入付费行情源。对于绝大多数个人项目，“准实时”足够用了。

4.3 第三步：加一个定时循环，让程序自己盯着盘

写完了单次获取，下一步自然就是循环。我推荐用schedule库或者直接用while True + time.sleep来做定时刷新，前者更适合复杂的定时策略，后者最直观且没有额外依赖。

我最常写的就是 while 循环版本，简单直接：

import time import datetime while True: now = datetime.datetime.now() # 只在交易时段内进行拉取，避免无意义请求 if now.weekday() < 5 and (9 <= now.hour <= 16): result = get_hk_quote_list(my_symbols) print(f"[{now.strftime('%H:%M:%S')}] 获取 {len(result)} 只股票数据") print(result) else: print("非交易时间，休眠中...") time.sleep(30) # 每30秒刷新一次

每次拉取的间隔要根据你的需求来定。如果只是做一个大盘观察，60 秒一次就够了；如果要做短线盯盘，建议 10 到 15 秒一次；再频繁的话，免费接口被限流的风险就上来了，而且对网络资源消耗也大。

4.4 第四步：把“拉数据”升级成“结构化数据管道”

实战中你会发现，光把数据打印到控制台没什么用，最终还是要落到存储和分析。我一般会用 pandas 做标准化处理后存入 CSV 或 SQLite，并在内存里维护近期数据的滚动窗口，方便触发条件判断。

我这里展示一个简化版的架构：

class HKQuoteMonitor: def __init__(self, symbols, interval=30): self.symbols = symbols self.interval = interval self.history = {} def fetch_once(self): df_list = [] for s in self.symbols: try: df = ak.stock_hk_hist_min_em(symbol=s, period="1") row = df.iloc[[-1]].copy() row["symbol"] = s df_list.append(row) except Exception as e: print(f"{s} error: {e}") if df_list: data = pd.concat(df_list, ignore_index=True) self._store(data) return data return pd.DataFrame() def _store(self, data): # 追加写入 CSV 或者写 SQLite，这里省略 pass

这个类的好处是把获取、存储和后续分析解耦了。后续你要加均线、加异动提醒、加微信通知，都是在_store或者新加一个方法里套一层逻辑的事，不会污染数据获取的核心代码。

4.5 第五步：异常处理与数据质量校验

在网络请求里，异常处理不是可选项，而是必需品。我在实际跑数据管道的场景里遇到过的异常类型有四五种，简单归纳一下：

网络层面：域名解析失败、连接超时、SSL 证书报错。这类异常通常表现为抛requests.exceptions系列异常，处理方式是捕获后重试，重试次数建议不超过三次。

接口层面的错误：上游数据源的接口地址改了、字段改了、返回结构变了。akshare 的接口如果遇到这类问题，往往会报KeyError或者解析失败。处理方式是捕获后换备用接口，而不是反复重试同一个已经坏掉的东西。

数据本身为空：接口返回了一个空 DataFrame，这种情况有可能是上游数据源暂时没有数据，也可能是因为传入了错误的代码格式。最简单的排查方法，是先用print(df)或者print(df.shape)确认返回数据的形态。

我写了一个带重试机制的通用请求函数：

import time import akshare as ak def safe_fetch(symbol, retries=3, wait=2): for i in range(retries): try: df = ak.stock_hk_hist_min_em(symbol=symbol, period="1") if df is None or df.empty: raise ValueError(f"{symbol} 返回空数据") return df except Exception as e: print(f"第 {i+1} 次请求 {symbol} 失败: {e}") if i < retries - 1: time.sleep(wait) raise RuntimeError(f"{symbol} 请求多次失败")

核心思想很简单：每次重试之前先睡一小会儿，避免在接口刚出问题的时候立刻把它打爆。重试次数不是越多越好，三次足够多，再多就是加重自身负担。

4.6 第六步：增量存储与去重策略

如果不做数据存储，那这个监控脚本的实时行情信息用一次就丢了。做策略分析的人一定有这个经验：历史数据永远不够用，所以抓到一条行情就存一条是必须的。

踩坑过的同学应该发现了，如果你每 30 秒请求一次分钟数据，同一分钟内请求两次拿到的其实是同一分钟的最新数据，按行追加会导致大量重复。常规解法是给 DataFrame 增加唯一键，比如用“代码 + 时间”作为主键，SQLite 写库时做INSERT OR REPLACE或者INSERT OR IGNORE，CSV 则先读后去重再写。

SQLite 是最适合个人项目的存储方式，不需要单独启动服务，文件即数据库，操作便捷。我用下面这段代码来完成去重入库：

import sqlite3 def init_db(db_path="hk_quote.db"): conn = sqlite3.connect(db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS quote_log ( symbol TEXT, dt TEXT, price REAL, volume REAL, PRIMARY KEY (symbol, dt) ) """) conn.commit() return conn def insert_quotes(conn, data): rows = [] for _, r in data.iterrows(): rows.append((r["symbol"], r["时间"], r["最新价"], r.get("成交量", 0))) conn.executemany( "INSERT OR REPLACE INTO quote_log (symbol, dt, price, volume) VALUES (?, ?, ?, ?)", rows, ) conn.commit()

这个方案跑一个月也就几十万行数据，SQLite 完全能扛住。等你的数据集到了几百万行级别，再考虑换 PostgreSQL 或者 MySQL 不迟。

5. 实战中必踩的坑：问题定位与效率优化

5.1 常见错误一览与快速定位

我把这半年写港股行情脚本过程中遇到的坑整理成一张表，按出现频率排序。这张表可以当排查手册用。

5.2 一个隐蔽的坑：分钟线接口的时间含义

stock_hk_hist_min_em这类分钟接口返回的每一行数据代表的是“这一分钟结束时的快照”，所以最后一行的“最新价”就是当前最新的价格。这一点要特别留意，因为你拿到的价格本身就有天然延迟，它是分钟聚合的，而不是每一笔成交的逐笔价。

如果你做的是逐笔级别的 Level 2 数据监控，免费接口做不到。但如果你只是做分钟级别的主题盯盘，这个粒度已经足够了。想清楚自己的需求层级，就不会被“实时”这个概念卡住。

5.3 提高运行效率的几个小技巧

第一个技巧：控制请求频率。不要每 5 秒拉一次全市场，这是免费接口最容易触发限流的用法。我建议个股接口轮询间隔不低于 10 秒，全市场快照接口不低于 30 秒。

第二个技巧：多线程不一定好。很多人一看到要监控很多票，就想到用多线程并发请求。但实际上如果监控的票少于 20 只，单线程按顺序请求完全够用，并发反而容易触发限流。如果你确实需要更高频次并发，那应该考虑换付费数据源。

第三个技巧：给网络请求加上超时时间。akshare 底层用的是 requests，很多接口没有显式设置 timeout，这导致上游数据源卡住时你的脚本也会无限期卡住。解决办法不是改 akshare 源码，而是在外层用进程级别的超时控制，比如下面的写法：

import concurrent.futures def fetch_with_timeout(symbol, timeout=15): with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(safe_fetch, symbol) try: return future.result(timeout=timeout) except concurrent.futures.TimeoutError: print(f"{symbol} 请求超时") return None

这个方案通用性很强，应对绝大多数卡死场景足够了。

5.4 数据准确性校验：别让脏数据干扰判断

有一次我发现腾讯控股的“最新价”是 309.5，但另一个数据源显示的是 310.2。这种差异是正常的，不同数据源的更新频率和聚合口径不同，并不代表谁一定是错的。

不过有些情况需要警惕：比如返回的价格突然变成 0，或者涨跌幅超过合理范围。这时候要做的不是相信数据，而是先做数据清洗。我一般在拿到数据后加一个过滤条件：

valid = df[(df["最新价"] > 0) & (df["成交量"] >= 0)]

用这个过滤后的数据去做计算，才能保证策略判断的基础是干净的。

提示：如果把实时行情用于自动交易决策，一定要做交叉验证。可以用两个不同数据源同时获取数据并做比对，偏差超过设定阈值则放弃本次交易信号。这是个人量化交易的基本安全措施，做主动交易的同学务必考虑。

6. 进阶扩展方向：从简单展示到可用系统

6.1 加一个“异动提醒”逻辑

数据管道跑通以后，下一步自然就是加业务逻辑。最典型的场景是价格异动提醒：当某只股票涨跌幅超过设定阈值时，自动推送消息到微信或钉钉。

示例逻辑可以这样写：

def check_alert(latest_price, prev_price, threshold=0.03): change = (latest_price - prev_price) / prev_price return change >= threshold, change prev = None while True: df = fetch_quote("h00700") price = df["最新价"].iloc[-1] if prev is not None: hit, pct = check_alert(price, prev) if hit: send_wechat_msg(f"腾讯控股 当前涨幅 {pct:.2%}") prev = price time.sleep(30)

推送用什么方式取决于你的环境。最简单的用pushplus、Server酱这类工具微信推送，一条 HTTP 请求就能完成。如果想彻底本地可控，可以直接写邮件通知，逻辑更简单。

6.2 把脚本升级成常驻服务

写好了功能，如果每次要手动跑到终端里执行，总是差点意思。我建议把它做成系统服务，启动后自动在后台运行。

Linux 下推荐用 systemd；macOS 可以用 launchd；Windows 下可以用任务计划程序或者 NSSM。给 systemd 一个最短示例：

[Unit] Description=HK Quote Monitor After=network.target [Service] ExecStart=/usr/bin/python3 /path/to/monitor.py WorkingDirectory=/path/to Restart=always [Install] WantedBy=multi-user.target

保存到/etc/systemd/system/hk-quote.service，然后执行：

sudo systemctl daemon-reload sudo systemctl enable --now hk-quote

这一步做完，你的行情监控就变成一个正式的常驻服务了，开机自启，异常自动重启，不用再操心进程挂掉的问题。

6.3 从“实时快照”到“历史数据回测”的桥接

很多人问：我有了实时行情，能拿来直接做回测吗？答案是：不适合。原因在于，实时行情接口只给你当前市场状态的一个截面，回测要的是连绵不断的完整时间序列。

更好的做法是：用实时行情脚本长期积累数据，攒一段时间后再用这些行情历史数据做回测。这相当于自己建了一个微型数据仓库。这个过程越久，数据越有价值。很多做个人量化的人，会专门把行情采集脚本挂在云服务器上跑几个月，就为了积累高质量的第一手数据。

如果你实在等不及自己攒数据，可以用ak.stock_hk_hist这类历史数据接口做分钟或日线级别的回测，但要注意数据的复权情况和精度，不同接口返回的历史数据可能因为复权方式不同而有差异。

7. 一些只有上手后才会懂的经验

代码写得再好，踩坑经验才是真正值钱的部分。最后分享几条我自己的心得体会。

第一条：先跑通最小闭环，再考虑系统化。不要一开始就想着要做什么监控大屏、数据库集群，先用十行代码把一只股票的行情打到屏幕上，确认链路通了，再慢慢扩展。我见过很多入门者栽在“想得太大，动手太小”上。

第二条：免费接口要当作不稳定资源来用。akshare 这种开源库，本质上是从公开数据源获取数据，上游一变动，接口就可能临时不好用。所以设计程序时一定要做好“备用接口”和“异常兜底”这两件事。数据源不可用的时候，程序宁可不更新，也不能带着错误数据往下跑。

第三条：对“实时”保持合理的预期。免费接口的“实时”是相对的，不是绝对的。如果你的交易策略是秒级甚至毫秒级的敏感策略，免费接口的数据延迟会直接导致策略失真。这种场景下，花点钱买合规的行情数据源才是最合理的投入。

第四条：数据积累从第一天就开始做。哪怕你现在的目标只是给朋友写个自动报价工具，也建议顺手把数据存下来。一个月后回头看，这些数据是比代码本身更有价值的资产。我自己最初那个每天跑 8 小时的采集脚本，现在已经成为本地量化研究里最基础的部分。

这个项目看起来做的是一个很小的技术点，但顺着“数据获取 → 数据处理 → 存储 → 监控提醒 → 积累历史”这条线往下走，你会发现它天然长出了一个完整数据管道的雏形。拿到实时行情只是第一步，怎么用好这些数据才是真正拉开差距的地方。