量化交易中的 Universe：股票池定义、过滤规则与数据获取
Universe 是什么 量化交易里，Universe（投资宇宙 / 股票池）指策略允许选股和交易的资产集合。简单理解： Universe = 你的策略在哪些股票里选比如：沪深 300 成分股（300 只） 中证 500 成分股（500 只） 全 A 股剔除 ST、停牌、次新股后的剩余标的 某个行业的所有上市公司为什么要定义 Universe A 股有 5000+ 只股票。如果不做筛选，直接在全市场跑因子计算：计算量大，速度慢 流动性差的股票出信号无法成交 次新股、ST 股信号质量差，噪声多先定义好 Universe，再在 Universe 里做因子计算、排名、选股，是量化策略的标准流程： 全市场（5000+只） ↓ Universe 过滤 符合条件的股票池（如 3000 只） ↓ 因子计算 每只股票打分 ↓ 排序选股 前 N 名 → 下单交易静态 Universe 直接用指数成分股，每次调仓时更新成分股列表： import akshare as ak# 沪深 300 成分股 df = ak.index_stock_cons_csindex(symbol="000300") universe = df["成分券代码"].tolist()# 中证 500 df = ak.index_stock_cons_csindex(symbol="000905") universe = df["成分券代码"].tolist()动态 Universe 每天收盘后重新计算，根据流动性、市值等条件动态生成： import pandas as pddef build_universe(daily_data: pd.DataFrame) -> list: """ daily_data: 包含 code, close, volume, market_cap, is_st, list_days 等列 """ universe = daily_data[ (~daily_data["is_st"]) & # 剔除 ST (daily_data["list_days"] >= 60) & # 上市满 60 天 (daily_data["volume"] > 0) & # 非停牌 (daily_data["market_cap"] > 5e9) & # 市值 > 50 亿 (daily_data["turnover_20d"] > 5e7) # 20 日均成交额 > 5000 万 ] return universe["code"].tolist()常用 Universe 类型Universe 说明 适合策略沪深 300 成分股 蓝筹龙头，流动性好 低换手率、稳健因子中证 500 成分股 中盘股，成长性较好 中小盘因子中证 1000 成分股 小盘股，弹性大 小盘动量、题材全 A 剔除 ST/停牌 覆盖面广 全市场扫描行业 Universe 限定某一行业 行业轮动、配对交易在量化框架中使用 JoinQuant（聚宽）： from jqdata import get_index_stocksuniverse = get_index_stocks("000300.XSHG") # 沪深 300Qlib： # qlib 配置文件 market: csi300vn.py / 自研框架： universe = build_universe(daily_df)for code in universe: factor_score = compute_factor(code) ...Universe 与 AI Agent 量化 在 AI Agent 驱动的量化系统里，Universe 还有另一层含义：限定 Agent 的分析范围。 universe: - A股 - 港股 - ETF配置好 Universe 后，Agent 的新闻抓取、因子计算、持仓分析都限制在这个范围内，避免无限制扩散导致计算量和 Token 消耗失控。
Python JSONDecodeError 排查：NaN、不完整结构、数字字符串的常见坑
JSONDecodeError 定位 json.loads() 报错时，JSONDecodeError 对象携带精确位置信息： import jsontry: data = json.loads(content) except json.JSONDecodeError as e: print(f"错误信息: {e.msg}") print(f"行号: {e.lineno}, 列号: {e.colno}") print(f"字符偏移: {e.pos}") print(f"出错内容: {content[max(0, e.pos-50):e.pos+50]}")原因一：裸 NaN / Infinity JSON 规范不允许 NaN 和 Infinity 作为数值字面量，只有加了引号的字符串 "NaN" 才合法： # 合法 json.loads('{"value": "NaN"}') # OK，返回字符串# 非法 json.loads('{"value": NaN}') # JSONDecodeError: Expecting value json.loads('{"value": Infinity}') # JSONDecodeError如果数据源返回了裸 NaN，需要预处理： import recleaned = re.sub(r'\bNaN\b', '"NaN"', content) data = json.loads(cleaned)原因二：内容不是完整 JSON 常见错误是将多个 JSON 对象直接拼接（不用数组包裹）： # 非法：两个对象并排 content = '{"a": 1}, {"b": 2}' json.loads(content) # JSONDecodeError# 合法：用数组包裹 content = '[{"a": 1}, {"b": 2}]' json.loads(content) # OK另一种情况是只收到了响应的一部分（网络截断），pos 错误位置通常在字符串末尾。 原因三：数字存为字符串 JSON 中数字加了引号，解析后得到字符串而非数值： {"x": "1228.8650628969294", "z": "NaN"}data = json.loads(content) type(data["x"]) # str，不是 float# 需要手动转换 x = float(data["x"])# z 是 "NaN" 字符串，转 float 得到 nan z = float(data["z"]) # float('nan') import math math.isnan(z) # True原因四：编码问题 非 UTF-8 编码的文件用 json.loads 读取会报错： # 读取文件时指定编码 with open('data.json', 'r', encoding='utf-8') as f: data = json.load(f)# 如果来源是 bytes data = json.loads(raw_bytes.decode('utf-8'))快速调试模板 import json, tracebackdef safe_parse(content: str): try: return json.loads(content), None except json.JSONDecodeError as e: context_start = max(0, e.pos - 30) context_end = min(len(content), e.pos + 30) snippet = content[context_start:context_end] marker = ' ' * (e.pos - context_start) + '^' print(f"JSONDecodeError at line {e.lineno}, col {e.colno}:") print(snippet) print(marker) return None, e
Python JSONDecodeError 排查清单：从 NaN 到编码 BOM
json.loads 抛 JSONDecodeError 时，异常信息里有行号、列号、字节偏移量——先把这三个印出来，问题基本就定位了： import jsontry: data = json.loads(content) except json.JSONDecodeError as e: print(e) print("line:", e.lineno, "col:", e.colno, "pos:", e.pos) print(content[max(0, e.pos - 40): e.pos + 40])最后一行把出错位置前后 40 个字符打出来，肉眼就能看出问题。以下是几种常见踩坑。 1. 根本不是合法 JSON（最常见） 抓了一段"对象序列"贴进来： { "id": 1 }, { "id": 2 }这不是合法 JSON，缺个数组外壳。要么改成： [ { "id": 1 }, { "id": 2 } ]要么按字段包一层： { "objects": [...] }2. NaN / Infinity 没引号 标准 JSON 不支持 NaN、Infinity。这样合法： {"z": "NaN"} ← 字符串这样非法： {"z": NaN} ← Python json.loads 会崩服务端如果是 Python json.dumps(..., allow_nan=True) 输出的，可能就是裸 NaN。解决办法： import json data = json.loads(content, parse_constant=lambda x: None) # 或 data = content.replace("NaN", "null") data = json.loads(data)3. 结尾多逗号 也是非标准 JSON： { "a": 1, "b": 2, ← 这个逗号 Python 不接受 }用 json5 或者预处理正则去掉最后一个逗号： import re content = re.sub(r",\s*([}\]])", r"\1", content)4. UTF-8 BOM 从 Windows 记事本另存的 JSON 常带 BOM，第一个字符是 ： json.loads(content) # JSONDecodeError: Expecting value: line 1 column 1 (char 0)读文件时用 utf-8-sig： with open("data.json", encoding="utf-8-sig") as f: data = json.load(f)5. 数字全被当字符串了 不会抛异常，但会让下游算错： {"x": "1228.86"}type(obj["x"]) 是 str，要计算得先转： x = float(obj["x"])批量转的话用一个小函数： def to_float_fields(d, fields): for f in fields: if f in d: d[f] = float(d[f]) return d6. 单引号不是 JSON 从 Python print(dict) 复制的字符串是单引号： {'id': 1, 'name': 'Tom'}这不是 JSON，是 Python repr。要么请对方 json.dumps 输出，要么本地 ast.literal_eval： import ast data = ast.literal_eval("{'id': 1, 'name': 'Tom'}")literal_eval 只解析字面量，比 eval 安全得多。 一句话总结 JSONDecodeError 第一步永远是把 e.pos 附近的原文打出来。九成情况能在 30 秒内定位到具体字符是啥。
Windows 查看端口被哪个进程占用：netstat / PowerShell / 资源监视器
cmd 方式（最常用） 打开命令提示符（管理员权限更好），查看指定端口： netstat -ano | findstr :9000输出示例： TCP 0.0.0.0:9000 0.0.0.0:0 LISTENING 12345末尾的 12345 是进程 PID，再查进程名： tasklist | findstr 12345输出： php-cgi.exe 12345 Console 1 25,200 Kphp-cgi.exe 就是占用 9000 端口的进程。 PowerShell 方式 直接查指定端口： Get-NetTCPConnection -LocalPort 9000输出包含 OwningProcess（PID），再查进程： Get-Process -Id (Get-NetTCPConnection -LocalPort 9000).OwningProcess这一条命令可以直接得到进程名，不需要中间步骤。 资源监视器（图形界面） 运行 resmon，进入「网络」→「监听端口」选项卡，直接看各进程占用的端口列表，也可以在「CPU」→「关联的句柄」里搜索端口号。 释放端口（结束进程） 拿到 PID 后，cmd： taskkill /F /PID 12345PowerShell： Stop-Process -Id 12345 -Force/F 或 -Force 表示强制终止，不等进程自行退出。 常见场景现象 原因启动服务报 Address already in use 端口被其他程序占用多个服务共用同一端口 需要改端口配置php-cgi / node / java 没停干净 上次启动的进程残留找到 PID 之后，先确认进程名再决定是否结束——别误杀系统进程。
用 Composer 从零搭建 PHP 框架：Router、IoC 容器、Middleware 到 ServiceProvider
自己搭一个最小 PHP 框架是理解 Laravel 架构的最好方式，比直接读 Laravel 源码有效得多。 项目结构 my-framework/ ├── app/ │ └── Controller/ │ └── HomeController.php ├── public/ │ └── index.php ├── src/ │ ├── Router.php │ ├── Container.php │ └── Middleware.php ├── vendor/ └── composer.json第一步：Composer 初始化 mkdir my-framework && cd my-framework composer initcomposer.json： { "name": "me/framework", "autoload": { "psr-4": { "App\\": "app/", "Framework\\": "src/" } } }生成自动加载文件： composer dump-autoload第二步：最小 Router src/Router.php： <?php namespace Framework;class Router { private array $routes = []; public function get(string $uri, $action): void { $this->routes['GET'][$uri] = $action; } public function post(string $uri, $action): void { $this->routes['POST'][$uri] = $action; } public function dispatch(): void { $method = $_SERVER['REQUEST_METHOD']; $uri = parse_url($_SERVER['REQUEST_URI'], PHP_URL_PATH); if (isset($this->routes[$method][$uri])) { call_user_func($this->routes[$method][$uri]); return; } http_response_code(404); echo "404 Not Found"; } }入口文件 public/index.php： <?php require '../vendor/autoload.php';use Framework\Router; use App\Controller\HomeController;$router = new Router(); $router->get('/', [HomeController::class, 'index']); $router->dispatch();启动： php -S localhost:8000 -t public第三步：IoC 容器（依赖注入） Laravel 的核心是 Container，不是 Router。 src/Container.php： <?php namespace Framework;use ReflectionClass;class Container { private array $bindings = []; public function bind(string $abstract, callable $factory): void { $this->bindings[$abstract] = $factory; } public function make(string $class): object { if (isset($this->bindings[$class])) { return ($this->bindings[$class])($this); } // 用反射自动解析构造函数依赖 $ref = new ReflectionClass($class); $constructor = $ref->getConstructor(); if (!$constructor) { return new $class(); } $params = array_map( fn($p) => $this->make($p->getType()->getName()), $constructor->getParameters() ); return $ref->newInstanceArgs($params); } }这样 $container->make(HomeController::class) 会自动实例化其所有依赖： class HomeController { public function __construct( private UserService $userService ) {} }第四步：Middleware 请求流程：Request → Middleware → Controller → Response <?php namespace Framework;class Middleware { private array $stack = []; public function add(callable $middleware): void { $this->stack[] = $middleware; } public function run(callable $handler): void { $runner = array_reduce( array_reverse($this->stack), fn($next, $mw) => fn() => $mw($next), $handler ); $runner(); } }示例认证中间件： $middleware->add(function (callable $next) { if (!isset($_SESSION['user'])) { http_response_code(401); echo '请先登录'; return; } $next(); });第五步：ServiceProvider <?php namespace Framework;abstract class ServiceProvider { public function __construct( protected Container $app ) {} abstract public function register(): void; }示例： class DatabaseServiceProvider extends ServiceProvider { public function register(): void { $this->app->bind('db', fn($app) => new Database( config('database.host'), config('database.name') )); } }注册： (new DatabaseServiceProvider($container))->register();学习路线Router + 自动加载 → 理解 HTTP 分发 Container + ReflectionClass → 理解依赖注入 Middleware → 理解 Laravel Pipeline ServiceProvider → 理解 Laravel 服务注册 自己实现后再看 Laravel 源码，对照理解完成前 3 步后基本就能看懂 Laravel 入口流程了。