feat: improve multi-source news output

Author: AdvancingTitans

CHANGELOG.md

@@ -5,6 +5,16 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.1.7] - 2026-06-01
+
+### Added
+- Added `young news <symbol>` for a quick single-stock news/momentum check without printing the full quote report.
+
+### Changed
+- News output now shows the source and link status on every item, including a clear "no public link" label instead of blank lines.
+- Multi-source news ranking now keeps per-source hit counts and uses a source-balanced display so Futu does not automatically crowd out Sina Finance or Eastmoney items when those sources have matching news.
+- Hong Kong stock news aliases now strip suffixes such as `-W` / `－Ｗ`, improving matches for names like Meituan-W.
+
 ## [0.1.6] - 2026-06-01
 
 ### Fixed

README.md

@@ -31,6 +31,7 @@ young hk                # Hong Kong indices snapshot
 young us                # US indices snapshot
 young global            # A + HK + US in one view
 young stock 600519      # one stock snapshot (A-share / HK / US)
+young news 3690.HK      # multi-source news only
 young stock AAPL --no-news
 young us --no-news      # market data only, skip news links
 young indices           # A-share indices only
@@ -68,6 +69,7 @@ It is also a foundation for analysis pipelines: every subcommand maps to a Pytho
 
 - **Multiple public quote sources** — Tencent Finance, Sina Finance, and Eastmoney are tried in sequence so temporary source failures can be filled by another no-login endpoint.
 - **Single-stock lookup** — `young stock 600519`, `young stock 0700.HK`, or `young stock AAPL` prints a compact quote snapshot with source, trade date, price, change, volume, turnover when available, and optional news.
+- **Single-stock news** — `young news 3690.HK` prints only the news/momentum view, with each item showing source and link status.
 - **Smart caching** — `~/.young_stock/cache/`, 7-day TTL, auto-pruned. Pass `--refresh` to skip.
 - **Trade-day awareness** — nearest-trade-day resolution including weekends and (best-effort) holidays.
 - **Rich terminal tables** — readable on dark and light terminals.
@@ -119,6 +121,7 @@ A 股盘后行情命令行工具。免登录、免 API key、免反爬技巧 —
 pip install young-stock-cli
 young a            # A 股盘后总览（主命令）
 young stock 600519 # 单只股票速览（A股 / 港股 / 美股）
+young news 3690.HK # 单只股票消息面，多源新闻与链接
 young hk --no-news # 只看行情，跳过新闻链接
 young zt-pool      # 涨停 / 跌停 / 炸板分析
 young flow         # 最新可核验 A 股资金流向

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "young-stock-cli"
-version = "0.1.6"
+version = "0.1.7"
 description = "A-share (China stock market) after-hours CLI — no login, no scraping tricks, just data."
 readme = "README.md"
 requires-python = ">=3.10"

src/young_stock/__init__.py

@@ -1,3 +1,3 @@
 """young-stock-cli: A-share after-hours CLI."""
 
-__version__ = "0.1.6"
+__version__ = "0.1.7"

src/young_stock/_core.py

@@ -1665,6 +1665,17 @@ def _clean_news_title(title: str) -> str:
     return html.unescape(title).strip()
 
 
+def _news_aliases(symbol: str, name: str = "") -> list[str]:
+    aliases = [symbol]
+    if symbol.upper().endswith(".HK"):
+        raw = symbol.upper().replace(".HK", "")
+        aliases.extend([raw, raw.lstrip("0"), raw.zfill(5)])
+    if name:
+        aliases.append(name)
+        aliases.append(re.split(r"[-－—（(]", name, maxsplit=1)[0].strip())
+    return [alias for alias in dict.fromkeys(a for a in aliases if a)]
+
+
 def _normalize_futu_feed(
     feed_data: dict[str, Any],
     size: int = 5,
@@ -1790,6 +1801,67 @@ def _parse_news_time(value: Any) -> int:
     return 0
 
 
+def _news_source_label(source: str) -> str:
+    return {
+        "futu_news": "富途资讯",
+        "futu_feed": "富途社区/资讯",
+        "sina_roll": "新浪财经",
+        "eastmoney_fast": "东方财富快讯",
+        "东方财富": "东方财富",
+        "新浪财经": "新浪财经",
+    }.get(source, source or "未知来源")
+
+
+def _normalize_news_url(item: dict[str, Any]) -> str:
+    for key in ("url", "jump_url", "link", "news_url", "article_url", "wapurl"):
+        value = item.get(key)
+        if value:
+            return str(value)
+    return ""
+
+
+def _select_diverse_news(items: list[dict[str, Any]], size: int) -> list[dict[str, Any]]:
+    """展示时尽量保留多来源；热度计算仍基于所有命中。"""
+    if len(items) <= size:
+        return items
+    buckets: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for item in items:
+        buckets[str(item.get("source", "news"))].append(item)
+    for bucket in buckets.values():
+        bucket.sort(key=lambda item: _parse_news_time(item.get("publish_time")), reverse=True)
+
+    selected: list[dict[str, Any]] = []
+    seen_titles: set[str] = set()
+    source_order = sorted(
+        buckets,
+        key=lambda source: (
+            -len(buckets[source]),
+            -max((_parse_news_time(item.get("publish_time")) for item in buckets[source]), default=0),
+        ),
+    )
+    while len(selected) < size and source_order:
+        progressed = False
+        for source in list(source_order):
+            bucket = buckets[source]
+            while bucket:
+                item = bucket.pop(0)
+                title = str(item.get("title", ""))
+                if title in seen_titles:
+                    continue
+                selected.append(item)
+                seen_titles.add(title)
+                progressed = True
+                break
+            if not bucket:
+                source_order.remove(source)
+            if len(selected) >= size:
+                break
+        if not progressed:
+            break
+    selected.sort(key=lambda item: _parse_news_time(item.get("publish_time")), reverse=True)
+    return selected[:size]
+
+
 def news_search_chain(keyword: str, size: int = 5, lang: str = "en", aliases: list[str] | None = None) -> dict[str, Any]:
     """资讯四段式：Futu 新闻 → Futu feed → 新浪财经 → 东方财富。"""
     news = futu_news_search(keyword, size=size, lang=lang)
@@ -1814,15 +1886,22 @@ def news_search_chain(keyword: str, size: int = 5, lang: str = "en", aliases: li
 
 def combined_news_search(keyword: str, size: int = 5, lang: str = "zh-CN", aliases: list[str] | None = None) -> dict[str, Any]:
     """聚合免登录新闻源，用于热度排序和展示。"""
+    per_source_size = max(size, 8)
     sources = [
-        futu_news_search(keyword, size=size, lang=lang),
-        _normalize_futu_feed(futu_stock_feed(keyword, size=size * 2), size=size, keyword=keyword, aliases=aliases),
-        sina_roll_news(keyword, size=size, aliases=aliases),
-        eastmoney_fast_news(keyword, size=size, aliases=aliases),
+        futu_news_search(keyword, size=per_source_size, lang=lang),
+        _normalize_futu_feed(
+            futu_stock_feed(keyword, size=per_source_size * 2),
+            size=per_source_size,
+            keyword=keyword,
+            aliases=aliases,
+        ),
+        sina_roll_news(keyword, size=per_source_size, aliases=aliases),
+        eastmoney_fast_news(keyword, size=per_source_size, aliases=aliases),
     ]
     seen: set[str] = set()
     items: list[dict[str, Any]] = []
     used_sources: list[str] = []
+    source_counts: Counter[str] = Counter()
     for idx, source in enumerate(sources):
         if not source.get("data"):
             continue
@@ -1836,10 +1915,19 @@ def combined_news_search(keyword: str, size: int = 5, lang: str = "zh-CN", alias
             seen.add(title)
             normalized = dict(item)
             normalized["title"] = title
+            normalized["url"] = _normalize_news_url(normalized)
             normalized.setdefault("source", source_name)
             items.append(normalized)
+            source_counts[source_name] += 1
     items.sort(key=lambda item: _parse_news_time(item.get("publish_time")), reverse=True)
-    return {"source": "+".join(s for s in used_sources if s) or "none", "data": items[:size]}
+    linked_items = [item for item in items if item.get("url")]
+    display_items = linked_items if len(linked_items) >= size else items
+    return {
+        "source": "+".join(s for s in used_sources if s) or "none",
+        "data": _select_diverse_news(display_items, size),
+        "all_count": len(items),
+        "source_counts": dict(source_counts),
+    }
 
 
 def rank_symbols_by_news_heat(
@@ -1859,9 +1947,10 @@ def rank_symbols_by_news_heat(
             aliases.append(names[symbol])
         keyword = names.get(symbol) or symbol
         news = combined_news_search(keyword, size=5, lang=lang, aliases=aliases)
-        score = 0.0
+        score = float(news.get("all_count", 0))
+        source_counts = news.get("source_counts") or {}
+        score += max(0, len(source_counts) - 1) * 0.8
         for item in news.get("data", []):
-            score += 1.0
             ts = _parse_news_time(item.get("publish_time"))
             if ts and now_ts - ts < 24 * 3600:
                 score += 0.5
@@ -2238,41 +2327,67 @@ def run_stock_quote(symbol: str, date_str: str, include_news: bool = True) -> No
     if include_news:
         keyword = qd.name if qd.market in ("cn_market", "hk_market") and qd.name else qd.symbol
         lang = "zh-CN" if qd.market in ("cn_market", "hk_market") else "en"
-        aliases = [qd.symbol, qd.name] if qd.name else [qd.symbol]
+        aliases = _news_aliases(qd.symbol, qd.name)
         print_futu_news(combined_news_search(keyword, size=5, lang=lang, aliases=aliases), keyword)
 
     if DIAGNOSTICS:
         print_diagnostic_summary()
     print_report_footer()
 
 
-def print_futu_news(news_data: dict, keyword: str) -> None:
+def run_stock_news(symbol: str, date_str: str, size: int = 8) -> None:
+    DIAGNOSTICS.clear()
+    normalized = symbol
+    market = detect_market_type(symbol)
+    name = ""
+    try:
+        normalized, market = normalize_stock_symbol(symbol)
+        qd = get_single_stock_quote(normalized, date_str)
+        if qd:
+            name = qd.name
+            market = qd.market
+    except ValueError:
+        pass
+
+    keyword = name if market in ("cn_market", "hk_market") and name else normalized
+    lang = "zh-CN" if market in ("cn_market", "hk_market") else "en"
+    aliases = _news_aliases(normalized, name)
+    print(f"# 消息面速览：{name or normalized} ({normalized})\n")
+    print_stage_line(date_str)
+    news = combined_news_search(keyword, size=size, lang=lang, aliases=aliases)
+    print_futu_news(news, keyword, limit=size)
+    if not news.get("data"):
+        print("暂未从免登录来源拿到匹配新闻，可以稍后加 --refresh 重试。")
+    print_report_footer()
+
+
+def print_futu_news(news_data: dict, keyword: str, limit: int = 5) -> None:
     if "_error" in news_data:
         return
     data = news_data.get("data", [])
     if not data:
         return
     source = news_data.get("source", "futu_news")
-    if "+" in str(source):
-        source_label = "多源聚合"
-    else:
-        source_label = {
-            "futu_news": "富途资讯",
-            "futu_feed": "富途社区/资讯",
-            "sina_roll": "新浪财经",
-            "eastmoney_fast": "东方财富快讯",
-        }.get(source, str(source))
-    print(f"## {keyword} 相关新闻（{source_label}，前5条）\n")
-    for i, item in enumerate(data[:5], 1):
+    source_label = "多源聚合" if "+" in str(source) else _news_source_label(str(source))
+    shown = min(limit, len(data))
+    print(f"## {keyword} 相关新闻（{source_label}，前{shown}条）\n")
+    source_counts = news_data.get("source_counts") or {}
+    if source_counts:
+        parts = [f"{_news_source_label(str(k))} {v}条" for k, v in source_counts.items()]
+        print("  来源覆盖: " + " / ".join(parts))
+        print()
+    for i, item in enumerate(data[:limit], 1):
         ts = item.get("publish_time", 0)
         # 富途返回的 publish_time 是字符串，先转 int
         try:
             ts_int = int(ts)
             dt_str = datetime.fromtimestamp(ts_int).strftime("%m-%d %H:%M") if ts_int else ""
         except (TypeError, ValueError):
             dt_str = ""
+        item_source = _news_source_label(str(item.get("source") or source))
+        url = _normalize_news_url(item)
         print(f"{i}. [{dt_str}] {_clean_news_title(item.get('title', ''))}")
-        print(f"   {item.get('url', '')}")
+        print(f"   来源: {item_source} | 链接: {url or '暂无公开链接'}")
     print()
 
 
@@ -2650,19 +2765,20 @@ def main():
             i += 1
         elif args[i] in ("--stock", "--symbol") and i + 1 < len(args):
             stock_symbol = args[i + 1]
-            market = "stock"
+            if market != "news":
+                market = "stock"
             i += 2
         elif re.fullmatch(r"\d{8}", args[i]):
             date_str = args[i]
             i += 1
         else:
             i += 1
 
-    if market not in ("a", "hk", "us", "global", "stock"):
-        print("错误: --market 参数必须是 a、hk、us、global 或 stock", file=sys.stderr)
+    if market not in ("a", "hk", "us", "global", "stock", "news"):
+        print("错误: --market 参数必须是 a、hk、us、global、stock 或 news", file=sys.stderr)
         sys.exit(1)
-    if market == "stock" and not stock_symbol:
-        print("错误: --market stock 需要配合 --stock 600519 使用", file=sys.stderr)
+    if market in ("stock", "news") and not stock_symbol:
+        print(f"错误: --market {market} 需要配合 --stock 600519 使用", file=sys.stderr)
         sys.exit(1)
 
     # 清理过期缓存
@@ -2681,6 +2797,8 @@ def main():
         run_global_market(date_str)
     elif market == "stock" and stock_symbol:
         run_stock_quote(stock_symbol, date_str, include_news=include_news)
+    elif market == "news" and stock_symbol:
+        run_stock_news(stock_symbol, date_str)
 
 
 if __name__ == "__main__":

src/young_stock/cli.py

@@ -136,6 +136,19 @@ def stock(symbol: str, date: str | None, refresh: bool, no_news: bool) -> None:
     _core.run_stock_quote(symbol, date_str, include_news=not no_news)
 
 
+@cli.command(help="Show multi-source news for one stock, e.g. 600519, 0700.HK, AAPL.")
+@click.argument("symbol")
+@_date_opt
+@_refresh_opt
+@click.option("--limit", default=8, show_default=True, help="Maximum news items to show.")
+def news(symbol: str, date: str | None, refresh: bool, limit: int) -> None:
+    if refresh:
+        _core.NO_CACHE = True
+    _core.cache_clear_old(days=7)
+    date_str = date or _core.nearest_trade_date()
+    _core.run_stock_news(symbol, date_str, size=limit)
+
+
 @cli.command(help="Clear cached responses older than N days.")
 @click.option("--days", default=7, show_default=True, help="Delete cache files older than this many days.")
 def cache_clear(days: int) -> None:

tests/test_cli.py

@@ -31,7 +31,7 @@ def test_cli_subcommands_registered():
     from click.testing import CliRunner
     runner = CliRunner()
     result = runner.invoke(cli, ["--help"])
-    for sub in ["a", "hk", "us", "global", "indices", "zt-pool", "flow", "stock", "cache-clear", "update"]:
+    for sub in ["a", "hk", "us", "global", "indices", "zt-pool", "flow", "stock", "news", "cache-clear", "update"]:
         assert sub in result.output, f"subcommand `{sub}` missing from help"
 
 
@@ -68,6 +68,21 @@ def test_cli_us_no_news(monkeypatch):
     assert calls == [("20260529", False)]
 
 
+def test_cli_news_runs_stock_news(monkeypatch):
+    from click.testing import CliRunner
+
+    calls = []
+    monkeypatch.setattr(cli_module._core, "nearest_trade_date", lambda: "20260529")
+    monkeypatch.setattr(cli_module._core, "cache_clear_old", lambda days: None)
+    monkeypatch.setattr(cli_module._core, "run_stock_news", lambda symbol, date_str, size=8: calls.append((symbol, date_str, size)))
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["news", "0700.HK", "--limit", "6"])
+
+    assert result.exit_code == 0
+    assert calls == [("0700.HK", "20260529", 6)]
+
+
 def test_cli_hk_no_news(monkeypatch):
     from click.testing import CliRunner