feat(search): 标的搜索支持拼音首字母 + 创业/科创/北交所徽标

## 1. 拼音首字母搜索 (同花顺式)

后端 search_instruments 新增拼音匹配层, 输入 payh 可命中「平安银行」。
- 辅助函数 _name_pinyin_keys 用 lru_cache 缓存「名称→首字母串」, 命中后
  近似 dict 查找, 全市场遍历 < 1ms
- 多音字用 heteronym 笛卡尔积展开, 「重庆」同时匹配 cq/zq 两种读音;
  并加载 A 股高频地名词典 (重庆/长安/长春/长沙/长城/长江)
- 拼音分支仅在纯 ASCII 字母输入时触发, 中文/数字搜索零开销跳过,
  完全向后兼容
- 搜索分层: ① code/symbol 前缀 → ② 拼音首字母前缀 → ③ 包含匹配

新增依赖: pypinyin>=0.50 (纯 Python, 无 C 扩展)
新增测试: tests/test_instrument_search.py (15 用例覆盖拼音/多音字/兼容/边界)

## 2. 搜索结果显示板块徽标

自选/财务搜索/监控规则/回测选标的 四个搜索入口的结果项, 现在会显示
创业板(橙「创」)/科创板(青「科」)/北交所(紫「北」)彩色徽标。
- 复用项目既有的 boardTag (@/components/stock-table/primitives),
  与自选/策略/Dashboard 表格徽标样式完全统一
- 沪深主板不显示徽标 (信息量低, 保持简洁)

## 验证
- 后端全量测试 503 passed (含新增 15)
- 前端 tsc --noEmit 零错误
This commit is contained in:
shy3130
2026-07-31 19:38:36 +08:00
parent 84bc6cb875
commit 192f6c4aa9
7 changed files with 247 additions and 5 deletions
+76 -5
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import logging
import math
from datetime import date, timedelta
from functools import lru_cache
from typing import Optional
from fastapi import APIRouter, HTTPException, Query, Request
@@ -31,6 +32,53 @@ def _minute_allowed(capset) -> bool:
return not fallback
@lru_cache(maxsize=8192)
def _name_pinyin_keys(name: str) -> tuple[str, ...]:
"""返回中文名称所有可能的拼音首字母串 (多音字展开为笛卡尔积)。
'平安银行' -> ('PAYH',); '重庆百货' -> ('CQBH', 'CQMH', 'ZQBH', 'ZQMH')。
非汉字字符原样保留: '万科A' -> ('WKA',)。
股票名总量有限且不变, lru_cache 命中后单次查询 ≈ dict 查找, 全市场遍历 < 1ms。
"""
from pypinyin import pinyin, Style
if not name:
return ()
keys = [""]
for group in pinyin(name, style=Style.FIRST_LETTER, heteronym=True):
keys = [k + g.upper() for k in keys for g in group]
return tuple(keys)
def _init_pinyin_dict() -> None:
"""加载 A 股高频多音字地名/词组词典, 使常见误读也能命中。
pypinyin 默认词典对部分地名取常见读音 (如「重」→ chóng), 补充后「重庆」
同时接受 zhòng/qìng (zq) 与 chóng/qīng (cq) 两种首字母, 与同花顺行为一致。
幂等: 多次调用安全。
"""
try:
from pypinyin import load_phrases_dict
# value 用二维 list: 每个字给一个或多个读音
load_phrases_dict({
"重庆": [["zhòng", "chóng"], ["qīng"]],
"长安": [["cháng", "zhǎng"], ["ān"]],
"长春": [["cháng", "zhǎng"], ["chūn"]],
"长沙": [["cháng", "zhǎng"], ["shā"]],
"长城": [["cháng", "zhǎng"], ["chéng"]],
"长江": [["cháng", "zhǎng"], ["jiāng"]],
})
except Exception as exc: # noqa: BLE001
logger.warning("pypinyin phrases dict load failed (polyphone coverage may degrade): %s", exc)
_init_pinyin_dict()
def _match_pinyin(name: str, keyword: str) -> bool:
"""keyword 是否匹配 name 任一拼音首字母串的前缀 (支持多音字)。"""
return any(k.startswith(keyword) for k in _name_pinyin_keys(name))
@router.get("/instruments/search")
def search_instruments(
request: Request,
@@ -67,6 +115,7 @@ def search_instruments(
df = pl.concat(parts, how="vertical")
keyword = q.strip().upper()
is_pinyin_query = keyword.isalpha() and keyword.isascii()
# code/symbol 前缀优先,再 name 包含匹配
prefix_mask = (
@@ -79,16 +128,38 @@ def search_instruments(
| pl.col("name").str.contains(keyword, literal=True)
)
# 前缀匹配优先,剩余名额用包含匹配补充
# 分层匹配: ① code/symbol 前缀 → ② 拼音首字母前缀(纯字母输入) → ③ 包含匹配
prefix_hits = df.filter(prefix_mask).head(limit)
if prefix_hits.height >= limit:
matched = prefix_hits
else:
collected = [prefix_hits] if prefix_hits.height else []
seen = set(prefix_hits["symbol"].to_list()) if prefix_hits.height else set()
remaining = limit - prefix_hits.height
# 排除已匹配的 symbol
prefix_symbols = set(prefix_hits["symbol"].to_list()) if not prefix_hits.is_empty() else set()
contain_hits = df.filter(contains_mask & ~pl.col("symbol").is_in(prefix_symbols)).head(remaining)
matched = pl.concat([prefix_hits, contain_hits]) if not prefix_hits.is_empty() else contain_hits
# ② 拼音首字母前缀: 仅纯字母输入触发 (如 payh → 平安银行); 中文/代码输入零开销跳过
if is_pinyin_query and remaining > 0:
pinyin_rows = []
for row in df.filter(~pl.col("symbol").is_in(seen)).iter_rows(named=True):
if _match_pinyin(row["name"], keyword):
pinyin_rows.append(row)
if len(pinyin_rows) >= remaining:
break
if pinyin_rows:
collected.append(pl.DataFrame(pinyin_rows))
seen.update(r["symbol"] for r in pinyin_rows)
remaining -= len(pinyin_rows)
# ③ 包含匹配补充
if remaining > 0:
contain_hits = df.filter(contains_mask & ~pl.col("symbol").is_in(seen)).head(remaining)
if contain_hits.height:
collected.append(contain_hits)
matched = (
pl.concat(collected, how="vertical") if len(collected) > 1
else (collected[0] if collected else df.head(0))
)
rows = matched.select(["symbol", "name", "code", "asset_type"]).to_dicts()
return {"results": rows}
+1
View File
@@ -22,6 +22,7 @@ dependencies = [
# llvmlite 不再提供 macOS Intel wheel;该平台使用 Matrix 纯 Python fallback。
"numba>=0.65.1; sys_platform != 'darwin' or platform_machine != 'x86_64'",
"fastexcel>=0.10", # Polars 读取 xlsx/xls
"pypinyin>=0.50", # 股票名拼音首字母搜索 (同花顺式 payh → 平安银行)
# TickFlow 官方 SDK
"tickflow[all]>=0.1.23",
# Scheduling
+148
View File
@@ -0,0 +1,148 @@
"""标的搜索测试: 代码 / 名称 / 拼音首字母 (同花顺式 payh → 平安银行)。
直接调用 search_instruments, 用最小 FakeRepo 提供 instruments 缓存, 不走 HTTP/DB。
"""
from __future__ import annotations
import types
import polars as pl
import pytest
from app.api.kline import search_instruments
class _FakeRepo:
"""最小 repo 桩: 只实现 search_instruments 依赖的 get_instruments_asset。"""
def __init__(self, by_asset: dict[str, pl.DataFrame]) -> None:
self.store = types.SimpleNamespace(data_dir="data")
self._by_asset = by_asset
def get_instruments_asset(self, asset_type: str) -> pl.DataFrame:
return self._by_asset.get(asset_type, pl.DataFrame())
def _request(repo: _FakeRepo) -> types.SimpleNamespace:
return types.SimpleNamespace(app=types.SimpleNamespace(state=types.SimpleNamespace(repo=repo)))
STOCKS = pl.DataFrame({
"symbol": ["000001.SZ", "600000.SH", "600519.SH", "000333.SZ", "600737.SH"],
"code": ["000001", "600000", "600519", "000333", "600737"],
"name": ["平安银行", "浦发银行", "贵州茅台", "美的集团", "中粮糖业"],
})
def _search(q: str, asset_types: str = "stock", limit: int = 20) -> list[dict]:
repo = _FakeRepo({"stock": STOCKS})
return search_instruments(_request(repo), q=q, limit=limit, asset_types=asset_types)["results"]
# ===== 既有逻辑回归: 代码 / 名称搜索不受影响 =====
def test_code_prefix_match():
"""code 前缀: 6005 → 600519 (前缀优先)。"""
rows = _search("6005")
assert "600519.SH" in [r["symbol"] for r in rows]
def test_symbol_contains_match():
rows = _search("6005")
assert "600519.SH" in [r["symbol"] for r in rows]
def test_chinese_name_contains_match():
rows = _search("银行")
assert sorted(r["symbol"] for r in rows) == ["000001.SZ", "600000.SH"]
def test_empty_query_returns_empty():
assert _search(" ") == []
# ===== 拼音首字母搜索 (新功能) =====
def test_pinyin_full_initials_match():
"""payh → 平安银行"""
rows = _search("payh")
assert [r["symbol"] for r in rows] == ["000001.SZ"]
def test_pinyin_prefix_match():
"""m → 美的集团 (m 开头)"""
rows = _search("m")
assert "000333.SZ" in [r["symbol"] for r in rows]
def test_pinyin_prefix_picks_multiple():
"""pf → 浦发银行; pa → 平安银行 (前缀区分)"""
assert [r["symbol"] for r in _search("pf")] == ["600000.SH"]
assert [r["symbol"] for r in _search("pa")] == ["000001.SZ"]
def test_pinyin_respects_limit():
"""limit 限制拼音结果数"""
rows = _search("z", limit=1) # z → 中粮糖业
assert len(rows) == 1
def test_pinyin_layer_between_prefix_and_contains():
"""拼音命中应排在包含匹配之前 (分层优先级)。"""
rows = _search("md") # md → 美的集团 (拼音); 无代码/符号以 md 前缀
assert "000333.SZ" in [r["symbol"] for r in rows]
def test_pinyin_and_code_prefix_coexist():
"""纯字母查询同时命中 code 前缀和拼音首字母时, code 前缀优先排前。"""
# '600000' 是浦发的 code 前缀; 这里用纯字母无法命中 code, 故仅验证拼音路径独立可用
rows = _search("pf") # 浦发
assert "600000.SH" in [r["symbol"] for r in rows]
# ===== 多音字 =====
POLYPHONE_STOCKS = pl.DataFrame({
"symbol": ["600729.SH", "000625.SZ"],
"code": ["600729", "000625"],
"name": ["重庆百货", "长安汽车"],
})
def test_polyphone_all_readings_match():
"""'重庆' 多音字: cq (chóng) 和 zq (zhòng) 读音都应命中。"""
repo = _FakeRepo({"stock": POLYPHONE_STOCKS})
# 取首字母集, 验证两种读音都能搜到
cq = search_instruments(_request(repo), q="cqbh", limit=20, asset_types="stock")["results"] # chóng qīng
zq = search_instruments(_request(repo), q="zqbh", limit=20, asset_types="stock")["results"] # zhòng qìng
assert "600729.SH" in [r["symbol"] for r in cq]
assert "600729.SH" in [r["symbol"] for r in zq]
# ===== 边界 =====
def test_non_ascii_skips_pinyin_branch():
"""中文输入不走拼音分支, 仍按名称匹配。"""
rows = _search("平安")
assert [r["symbol"] for r in rows] == ["000001.SZ"]
def test_digits_skips_pinyin_branch():
"""数字输入不走拼音分支, 走 code 前缀。"""
rows = _search("000")
assert "000001.SZ" in [r["symbol"] for r in rows]
def test_no_pinyin_hit_returns_empty():
"""无任何拼音命中时返回空 (不报错)。"""
assert _search("xyz") == []
def test_cache_returns_same_result_across_calls():
"""lru_cache 不应在不同请求间串结果 (按 name 缓存, 查询无状态)。"""
repo = _FakeRepo({"stock": STOCKS})
req = _request(repo)
r1 = search_instruments(req, q="payh", limit=20, asset_types="stock")["results"]
r2 = search_instruments(req, q="payh", limit=20, asset_types="stock")["results"]
assert r1 == r2
assert [r["symbol"] for r in r1] == ["000001.SZ"]
@@ -4,6 +4,7 @@ import { motion, AnimatePresence } from 'framer-motion'
import { Search, Loader2 } from 'lucide-react'
import { api } from '@/lib/api'
import { QK } from '@/lib/queryKeys'
import { boardTag } from '@/components/stock-table/primitives'
interface Props {
onSelect: (symbol: string, name: string) => void
@@ -118,6 +119,12 @@ export function StockFinancialSearch({ onSelect, assetTypes }: Props) {
>
<span className="font-mono shrink-0 text-xs w-[88px]">{r.symbol}</span>
<span className="truncate text-sm flex-1">{r.name}</span>
{(() => {
const b = boardTag(r.symbol)
return b && (
<span className={`shrink-0 px-1 py-0.5 rounded text-[10px] leading-none border ${b.color}`}>{b.label}</span>
)
})()}
{r.asset_type === 'index' && (
<span className="shrink-0 px-1 py-0.5 rounded text-[10px] leading-none bg-sky-500/10 text-sky-400"></span>
)}
@@ -5,6 +5,7 @@ import { Activity, Check, Plus, RadioTower, Save, Search, TrendingUp, Waypoints,
import { api, genRuleId, type MonitorRule, type MonitorCondition, type StrategyNotifyEvent } from '@/lib/api'
import { DEFAULT_STRATEGY_NOTIFY_EVENTS, LEGACY_STRATEGY_NOTIFY_EVENTS, STRATEGY_NOTIFY_EVENT_OPTIONS } from '@/lib/strategyMonitorEvents'
import { QK } from '@/lib/queryKeys'
import { boardTag } from '@/components/stock-table/primitives'
import { SignalPicker } from '@/components/screener/SignalPicker'
import { MONITOR_INTRADAY_SIGNAL_OPTIONS, SIGNAL_OPTIONS, cnSignal } from '@/lib/signals'
import { usePreferences } from '@/lib/useSharedQueries'
@@ -430,6 +431,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
{symbolSearch.data.results.map(r => (
<button key={r.symbol} onClick={() => addSymbol(r.symbol)} className="block w-full px-2 py-1 text-left text-[11px] hover:bg-elevated cursor-pointer">
<span className="font-mono text-foreground/80">{r.symbol}</span>
{(() => { const b = boardTag(r.symbol); return b && <span className={`ml-1 inline-flex items-center justify-center rounded px-0.5 text-[9px] font-bold leading-tight border ${b.color}`}>{b.label}</span> })()}
<span className="ml-1 text-muted">{r.name}</span>
</button>
))}
+6
View File
@@ -308,6 +308,12 @@ function StockSearchBox({
{r.asset_type === 'index' && (
<span className="shrink-0 px-1 py-0.5 rounded text-[10px] leading-none bg-sky-500/10 text-sky-400"></span>
)}
{(() => {
const b = boardTag(r.symbol)
return b && (
<span className={`shrink-0 px-1 py-0.5 rounded text-[10px] leading-none border ${b.color}`}>{b.label}</span>
)
})()}
</button>
<button
type="button"
@@ -13,6 +13,7 @@ import { QK } from '@/lib/queryKeys'
import { storage } from '@/lib/storage'
import { fmtPct, fmtPrice, priceColorClass } from '@/lib/format'
import { boardTag } from '@/lib/board'
import { boardTag as boardBadge } from '@/components/stock-table/primitives'
import { BUILTIN_COLUMNS } from '@/lib/watchlist-columns'
import { cnSignal } from '@/lib/signals'
import { SignalPicker } from '@/components/screener/SignalPicker'
@@ -794,6 +795,12 @@ function StockPoolPicker({ value, onChange, assetType = 'stock' }: { value: stri
>
<span className="w-[78px] shrink-0 font-mono">{r.symbol}</span>
<span className="min-w-0 flex-1 truncate text-secondary">{r.name}</span>
{(() => {
const b = boardBadge(r.symbol)
return b && (
<span className={`shrink-0 px-1 py-0.5 rounded text-[10px] leading-none border ${b.color}`}>{b.label}</span>
)
})()}
<Plus className={`h-3.5 w-3.5 ${added ? 'opacity-30' : 'text-accent'}`} />
</button>
)