--- name: uwillberich description: Build next-session A-share game plans from market structure, overnight macro shocks, policy timing, and watchlist leadership. Use when the user asks what A-shares may do tomorrow, which sectors may repair first, how to read the open, or wants a reusable pre-open discretionary decision workflow. metadata: {"openclaw":{"emoji":"📈","homepage":"https://github.com/huangrichao2020/uwillberich","requires":{"bins":["python3"]}}} --- # uwillberich Author: 超超 Contact: `grdomai43881@gmail.com` ## Overview Use this skill for decision-oriented A-share analysis. The goal is not to explain the market mechanically, but to convert today’s tape and overnight developments into a concrete next-session plan. Best fit: - next-session A-share outlook - likely repair sectors after a selloff - opening checklist for `09:00`, `09:25`, and `09:30-10:00` - first-30-minute observation template for distinguishing true repair from defensive concentration - watchlist-based decision notes - distinguishing defensive leadership from true market repair - persistent message iteration that maps high-attention news into watchlist overlays - automatic event-driven stock pools that feed directly into desk reports - main-force capital-flow confirmation for watchlists and market-wide risk tone - industry-chain expansion that turns event themes into fresh stock pools - sentiment scoring built from breadth, sector dispersion, and capital flow ## Core Workflow 1. Gather market structure first. - Confirm `EM_API_KEY` is configured before running any script. - Run `scripts/fetch_market_snapshot.py` for indices, breadth, and sector leaders/laggards. - Run `scripts/fetch_quotes.py` or `scripts/morning_brief.py` for the watchlist. 2. Confirm the overnight and policy layer. - Use primary sources first for `PBOC`, `Federal Reserve`, and other central-bank decisions. - Use high-quality news sources for geopolitics, oil, and global risk sentiment. 3. Classify the market through three layers. - External shock: oil, rates, U.S. equities, geopolitics - Domestic policy/liquidity: `LPR`, PBOC posture, macro support - Internal structure: breadth, leadership, relative strength, style rotation 4. Build a scenario tree. - Provide `Base / Bull / Bear` paths with explicit triggers and invalidations. 5. Turn the view into an execution checklist. - Include `09:00`, `09:20-09:25`, `09:30-10:00`, and `14:00-14:30`. ## Workflow Shortcuts - `Step 1: overnight and policy` - `scripts/mx_toolkit.py preset --name preopen_policy` - `scripts/mx_toolkit.py preset --name preopen_global_risk` - `Step 2: board resonance` - `scripts/fetch_market_snapshot.py` - `scripts/capital_flow.py` - `scripts/market_sentiment.py` - `scripts/mx_toolkit.py preset --name board_optical_module` - `scripts/mx_toolkit.py preset --name board_compute_power` - `Step 3: single-name validation` - `scripts/fetch_quotes.py` - `scripts/mx_toolkit.py preset --name validate_inspur` - `scripts/mx_toolkit.py preset --name validate_luxshare` - `Step 4: event-to-chain expansion` - `scripts/industry_chain.py` - `scripts/news_iterator.py` - `Source benchmark` - `scripts/benchmark_sources.py` ## Decision Heuristics - Prefer sectors that resisted best in a weak tape over sectors that merely fell the most. - Treat defensive leadership as separate from broad market repair. - On monthly `LPR` days, use the `09:00` release as a hard branch in the plan. - A repair thesis is stronger when leadership broadens from core growth names into secondary names and brokers. - A rebound without breadth is usually just a technical bounce. ## Scripts Use these scripts before writing the decision note: - `scripts/fetch_market_snapshot.py` - Pulls Eastmoney index and sector breadth data. - `scripts/fetch_quotes.py` - Pulls Tencent quote snapshots for user-specified names. - `scripts/morning_brief.py` - Builds a markdown brief from the default watchlists in `assets/default_watchlists.json`. - `scripts/capital_flow.py` - Pulls the whole-market main-force snapshot plus top inflow/outflow names and intersects them with watchlists. - `scripts/market_sentiment.py` - Scores the tape as `抱团行情`, `科技修复`, `修复扩散`, or `分化偏弱` using breadth, sector dispersion, and capital flow. - `scripts/opening_window_checklist.py` - Builds a first-30-minute observation sheet with time gates, group scoreboards, and watchlist signal tables. - `scripts/industry_chain.py` - Uses event summaries and desk groups to expand into industry-chain stock pools through live MX stock screens. - `scripts/news_iterator.py` - Continuously polls public RSS feeds, classifies high-attention events, maps them into watchlist overlays, and writes dynamic event-driven stock pools. - `scripts/runtime_config.py` - Loads local runtime credentials, enforces the required `EM_API_KEY`, and prints the Eastmoney application URL when it is missing. - `scripts/mx_toolkit.py` - Calls the live Meixiang / Eastmoney APIs for news search, stock screening, structured data queries, and preset desk workflows. - `scripts/benchmark_sources.py` - Benchmarks public and MX-enhanced sources before you decide what to trust as the primary feed. - `scripts/install_news_iterator_launchd.py` - Installs the news iterator as a `launchd` job on macOS for long-running local polling. - `scripts/smoke_test.py` - Verifies that the bundled scripts and public endpoints are working. ## References Read only what you need: - `references/methodology.md` - Trading philosophy, decision tree, and timing gates. - `references/data-sources.md` - Source map for official and market data endpoints. - `references/persona-prompt.md` - Decision-maker persona for desk-style answers. - `references/trading-mode-prompt.md` - Time-boxed opening workflow for the next A-share session. - `references/opening-window-template.md` - A reusable first-30-minute decision template. - `references/cross-cycle-watchlist.md` - How to use the cross-cycle core stock pool without turning it into an unfocused mega-list. - `references/event-regime-watchlists.md` - How to use war-shock and energy-spike watchlists as temporary overlays. - `references/message-iterator.md` - How to run the persistent RSS iterator, generate event-driven stock pools, and feed them into the desk workflow. - `assets/mx_presets.json` - Preset MX workflows for policy scan, global-risk scan, board resonance, and single-name validation. - `assets/industry_chains.json` - Theme-to-chain map for optical module, compute power, semiconductors, robotics, oil and coal, and IDC/power-cost overlays. ## Output Standard Default to a compact desk-style answer: - one-paragraph decision summary - `Base / Bull / Bear` path - most likely repair sectors - defensive-only sectors - opening checklist - `do / avoid` ## Required Credential - `EM_API_KEY` is mandatory for this skill. - Apply here: `https://ai.eastmoney.com/mxClaw` - After opening the link, click download and you will see the key. - Official site: `https://ai.eastmoney.com/nlink/` - Store it in `~/.uwillberich/runtime.env` FILE:README.md # uwillberich A ClawHub/OpenClaw-ready skill for next-session A-share discretionary planning. It is designed for one job: turn today’s tape and overnight developments into a concrete game plan for tomorrow’s open. Author: 超超 Contact: `grdomai43881@gmail.com` GitHub is the main source of truth for installation and updates: ```text https://github.com/huangrichao2020/uwillberich ``` ## Good Use Cases - "What is the most likely A-share path tomorrow?" - "Which sectors are most likely to repair first after today’s selloff?" - "Give me a `09:00 / 09:25 / 09:30-10:00` opening checklist." - "Build a watchlist-driven pre-open note for A-shares." - "Tell me whether this is real repair or just defensive concentration." - "Use the cross-cycle core stock pool to narrow tomorrow's key observation list." - "In a war-oil shock regime, tell me which A-share groups benefit and which ones get hurt." - "Continuously watch public news and map major events into A-share watchlists." - "Run a preset `Step 1 / Step 2 / Step 3` desk workflow and save all artifacts." - "Benchmark which public and MX data sources are healthy before the open." ## Workflow Map 1. `Step 1: overnight and policy` - `scripts/news_iterator.py` - `scripts/mx_toolkit.py preset --name preopen_policy` - `scripts/mx_toolkit.py preset --name preopen_global_risk` 2. `Step 2: board resonance` - `scripts/fetch_market_snapshot.py` - `scripts/morning_brief.py` - `scripts/capital_flow.py` - `scripts/market_sentiment.py` - `scripts/mx_toolkit.py preset --name board_optical_module` - `scripts/mx_toolkit.py preset --name board_compute_power` 3. `Step 3: single-name validation` - `scripts/fetch_quotes.py` - `scripts/mx_toolkit.py preset --name validate_inspur` - `scripts/mx_toolkit.py preset --name validate_luxshare` 4. `Step 4: chain expansion` - `scripts/industry_chain.py` - `scripts/news_iterator.py` - `scripts/opening_window_checklist.py` 5. `Source benchmark` - `scripts/benchmark_sources.py` ## What This Skill Contains - `SKILL.md`: main instructions and trigger description - `references/methodology.md`: decision framework - `references/data-sources.md`: primary and market data sources - `references/persona-prompt.md`: decision-maker persona prompt - `references/trading-mode-prompt.md`: time-based pre-open trading mode prompt - `references/cross-cycle-watchlist.md`: how to use the cross-cycle core stock pool - `references/event-regime-watchlists.md`: war-shock overlay watchlists - `references/message-iterator.md`: persistent message iterator for high-attention news - `assets/mx_presets.json`: preset `Step 1 / Step 2 / Step 3` MX workflows - `scripts/fetch_market_snapshot.py`: index and sector breadth snapshot - `scripts/fetch_quotes.py`: Tencent quote watchlist snapshot - `scripts/morning_brief.py`: one-command markdown morning brief - `scripts/capital_flow.py`: main-force capital-flow overlay for the market and watchlists - `scripts/market_sentiment.py`: breadth + board-dispersion + capital-flow sentiment classifier - `scripts/opening_window_checklist.py`: first-30-minute decision sheet - `scripts/industry_chain.py`: event-to-industry-chain expansion for fresh stock pools - `scripts/news_iterator.py`: RSS polling, classification, SQLite state, markdown/jsonl outputs, and automatic event stock pools - `scripts/runtime_config.py`: local credential helper for the required `EM_API_KEY` - `scripts/mx_api.py`: Meixiang / Eastmoney API wrapper for live finance queries - `scripts/mx_toolkit.py`: CLI wrapper for real news search, stock screen, structured data queries, and desk presets - `scripts/benchmark_sources.py`: source latency / availability benchmark - `scripts/install_news_iterator_launchd.py`: macOS launchd installer for scheduled polling - `scripts/smoke_test.py`: local smoke test for the bundled scripts ## Agent Install Install this folder into: - `~/.codex/skills/uwillberich` - `~/.openclaw/skills/uwillberich` Example: ```bash git clone https://github.com/huangrichao2020/uwillberich.git mkdir -p ~/.codex/skills cp -R uwillberich/skill/uwillberich ~/.codex/skills/uwillberich ``` One-line install for Codex: ```bash git clone https://github.com/huangrichao2020/uwillberich.git && cd uwillberich && ./install_skill.sh ``` One-line install for OpenClaw: ```bash git clone https://github.com/huangrichao2020/uwillberich.git && cd uwillberich && ./install_skill.sh openclaw ``` ## Keys And Credentials This skill hard-requires `EM_API_KEY`. - Apply here: `https://ai.eastmoney.com/mxClaw` - After opening the link, click download and you will see the key. - Official site: `https://ai.eastmoney.com/nlink/` - Store it locally in `~/.uwillberich/runtime.env`. - Check or set it with: ```bash python3 scripts/runtime_config.py status printf '%s' 'your_em_api_key' | python3 scripts/runtime_config.py set-em-key --stdin ``` Without `EM_API_KEY`, the scripts will exit and print the application URL plus setup command. - GitHub read access: only if the repo is private and an agent must clone it - GitHub write access: only if an agent should push changes back - Model-provider API keys: may be required by the host agent environment, but not by this skill itself ## Local Smoke Test ```bash python3 scripts/smoke_test.py python3 scripts/runtime_config.py status python3 scripts/mx_toolkit.py list-presets python3 scripts/mx_toolkit.py preset --name preopen_repair_chain python3 scripts/mx_toolkit.py preset --name flow_main_force python3 scripts/mx_toolkit.py news-search --query '立讯精密最新资讯' python3 scripts/mx_toolkit.py stock-screen --keyword 'A股光模块概念股' --page-size 10 --csv-out /tmp/cpo.csv --desc-out /tmp/cpo-columns.md python3 scripts/mx_toolkit.py query --tool-query '浪潮信息最新价市值' python3 scripts/capital_flow.py --groups tech_repair defensive_gauge python3 scripts/market_sentiment.py python3 scripts/industry_chain.py --groups tech_repair defensive_gauge python3 scripts/benchmark_sources.py python3 scripts/fetch_market_snapshot.py --format markdown python3 scripts/fetch_quotes.py sz300502 sh688981 sh600938 python3 scripts/morning_brief.py --groups core10 tech_repair python3 scripts/morning_brief.py --groups cross_cycle_anchor12 python3 scripts/morning_brief.py --groups cross_cycle_ai_hardware cross_cycle_semis cross_cycle_software_platforms cross_cycle_defense_industrial python3 scripts/morning_brief.py --groups war_shock_core12 python3 scripts/morning_brief.py --groups war_benefit_oil_coal war_headwind_compute_power python3 scripts/opening_window_checklist.py --groups tech_repair defensive_gauge policy_beta python3 scripts/news_iterator.py poll python3 scripts/news_iterator.py report --hours 12 python3 scripts/install_news_iterator_launchd.py install --interval-seconds 300 python3 scripts/morning_brief.py python3 scripts/opening_window_checklist.py ``` ## Optional ClawHub Publish From this folder: ```bash clawhub login clawhub publish /absolute/path/to/uwillberich --slug uwillberich --name "uwillberich" --version 0.1.7 --tags latest,finance,a-share,china,markets ``` ## Notes - ClawHub publishes a skill folder with `SKILL.md` plus supporting text files. - This skill uses only text-based resources and Python standard library scripts. - `EM_API_KEY` is mandatory for this skill. - The runtime helper automatically maps `EM_API_KEY` to the `MX_APIKEY` convention used by the public MX skills. - Preset and benchmark outputs default to `~/.uwillberich/data/`. - If `clawhub publish .` misreads the folder, use an absolute path or pass `--workdir` explicitly. - The opening-window script is intended for `09:00-10:00` use, especially the first 30 minutes after the A-share cash open. - For the larger quality pool, use `cross_cycle_anchor12` daily and reserve `cross_cycle_core` for weekly or phase-rotation review. - For geopolitical shocks, treat `war_benefit_oil_coal` and `war_headwind_compute_power` as temporary regime overlays, not permanent core watchlists. - If you only want one wartime overlay, start with `war_shock_core12`. - For continuous event intake, run `news_iterator.py` as a local service and treat the alert stream as an overlay, not a replacement for tape and breadth. - The morning brief and opening checklist can automatically append event-driven stock pools when `event_watchlists.json` exists in the default state directory. FILE:agents/openai.yaml # Author: 超超 # Contact: grdomai43881@gmail.com interface: display_name: "uwillberich" short_description: "A-share pre-open and opening workflow" brand_color: "#C1272D" default_prompt: "Use $uwillberich to build tomorrow's A-share game plan, likely repair sectors, and a first-30-minute opening checklist from overnight events and watchlists." policy: allow_implicit_invocation: true FILE:assets/default_watchlists.json { "core10": [ {"symbol": "sz300502", "name": "新易盛", "role": "CPO总龙头", "strong_signal": "9:45前稳在昨收上方并接近昨高，说明科技修复最真", "weak_signal": "跌回昨收下方并逼近昨低，说明CPO主线承接不足"}, {"symbol": "sz300308", "name": "中际旭创", "role": "光模块核心", "strong_signal": "跟随新易盛站上昨收且不破昨低", "weak_signal": "始终压在昨收下方，说明修复扩散不足"}, {"symbol": "sz000977", "name": "浪潮信息", "role": "算力服务器", "strong_signal": "站回昨收且量能跟随，说明算力链扩散", "weak_signal": "低开后无法翻红，说明修复只停留在CPO单点"}, {"symbol": "sh688981", "name": "中芯国际", "role": "半导体抗跌核心", "strong_signal": "稳在昨收上方并向昨高推进，说明半导体修复有质量", "weak_signal": "跌破昨低，说明成长风格仍弱"}, {"symbol": "sh688041", "name": "海光信息", "role": "高弹性科技", "strong_signal": "快速翻红并守住昨收，说明风险偏好回暖", "weak_signal": "继续走弱并失守昨低，说明高弹性仍被压制"}, {"symbol": "sh603986", "name": "兆易创新", "role": "半导体弹性票", "strong_signal": "站回昨收且不回踩昨低，说明半导体跟随性不错", "weak_signal": "冲高回落量大价弱，说明弹性票承接不足"}, {"symbol": "sz300059", "name": "东方财富", "role": "风险偏好代理", "strong_signal": "站上昨收并持续放量，说明情绪修复成立", "weak_signal": "红不了或冲高回落，说明风险偏好没有回来"}, {"symbol": "sh600030", "name": "中信证券", "role": "券商确认票", "strong_signal": "与东方财富同步转强，说明修复扩散到金融", "weak_signal": "始终弱于东方财富，说明机构态度仍偏谨慎"}, {"symbol": "sz000002", "name": "万科A", "role": "政策链检验票", "strong_signal": "LPR偏暖后站回昨收并冲击昨高，说明政策链被认可", "weak_signal": "政策后仍弱，说明资金不信地产链修复"}, {"symbol": "sh600938", "name": "中国海油", "role": "避险对照票", "strong_signal": "继续领涨，说明避险仍主导", "weak_signal": "回落而科技组转强，说明风格切回成长"} ], "tech_repair": [ {"symbol": "sz300502", "name": "新易盛", "role": "CPO"}, {"symbol": "sz300308", "name": "中际旭创", "role": "光模块"}, {"symbol": "sz000977", "name": "浪潮信息", "role": "算力"}, {"symbol": "sh688981", "name": "中芯国际", "role": "半导体"}, {"symbol": "sh688041", "name": "海光信息", "role": "高弹性AI"}, {"symbol": "sh603986", "name": "兆易创新", "role": "半导体弹性"} ], "policy_beta": [ {"symbol": "sz300059", "name": "东方财富", "role": "券商弹性"}, {"symbol": "sh600030", "name": "中信证券", "role": "券商确认"}, {"symbol": "sz000002", "name": "万科A", "role": "地产链检验"}, {"symbol": "sz000333", "name": "美的集团", "role": "家电", "strong_signal": "LPR或稳增长偏暖后率先翻红，说明消费地产链被接受", "weak_signal": "政策偏暖仍起不来，说明家电链跟随不足"}, {"symbol": "sh600048", "name": "保利发展", "role": "地产蓝筹", "strong_signal": "政策后快速站上昨收，说明地产链修复更可信", "weak_signal": "政策后仍弱，说明地产预期没有修复"} ], "defensive_gauge": [ {"symbol": "sh600938", "name": "中国海油", "role": "油气"}, {"symbol": "sh601857", "name": "中国石油", "role": "油气", "strong_signal": "继续领涨，说明避险和能源逻辑仍在", "weak_signal": "回落且成长转强，说明市场从防御切回进攻"}, {"symbol": "sh601088", "name": "中国神华", "role": "煤炭", "strong_signal": "继续走强，说明高股息防御仍占上风", "weak_signal": "走弱且科技修复，说明防御抱团松动"}, {"symbol": "sh600941", "name": "中国移动", "role": "电信", "strong_signal": "稳中偏强，说明低波红利仍被偏好", "weak_signal": "转弱而科技走强，说明风险偏好回升"}, {"symbol": "sh601398", "name": "工商银行", "role": "大行", "strong_signal": "大行继续强，说明避险和高股息主导", "weak_signal": "大行回落而券商回暖，说明风格切换"} ], "cross_cycle_anchor12": [ {"symbol": "sz002463", "name": "沪电股份", "role": "高速PCB龙头", "strong_signal": "先于硬件链翻红放量，说明高景气算力硬件回流", "weak_signal": "高开低走且弱于CPO，说明硬件链承接不足"}, {"symbol": "sz002475", "name": "立讯精密", "role": "精密制造龙头", "strong_signal": "稳步走强并带动果链，说明大票制造开始接力", "weak_signal": "始终水下，说明消费制造风格未修复"}, {"symbol": "sz300124", "name": "汇川技术", "role": "工控自动化龙头", "strong_signal": "稳在昨收上方，说明机构偏好回到硬核制造", "weak_signal": "转弱，说明高质量制造暂未获资金回流"}, {"symbol": "sz000063", "name": "中兴通讯", "role": "通信设备龙头", "strong_signal": "站回昨收并带动通信设备，说明基础设施链回暖", "weak_signal": "弱于服务器和光模块，说明修复仍不完整"}, {"symbol": "sz002230", "name": "科大讯飞", "role": "AI应用龙头", "strong_signal": "率先转强，说明AI应用方向开始补涨", "weak_signal": "持续疲弱，说明应用侧仍无资金承接"}, {"symbol": "sz000977", "name": "浪潮信息", "role": "AI服务器龙头", "strong_signal": "站回昨收且量能跟随，说明算力链扩散", "weak_signal": "持续水下，说明修复只停留在局部"}, {"symbol": "sh688041", "name": "海光信息", "role": "国产算力核心", "strong_signal": "快速翻红并守住昨收，说明风险偏好回暖", "weak_signal": "继续走弱，说明高弹性国产算力仍被压制"}, {"symbol": "sh603019", "name": "中科曙光", "role": "服务器平台核心", "strong_signal": "跟随海光和浪潮信息走强，说明平台链承接更完整", "weak_signal": "弱于上游芯片和CPO，说明平台链修复不够"}, {"symbol": "sh601138", "name": "工业富联", "role": "AI服务器制造", "strong_signal": "稳步走强，说明大容量AI硬件制造受认可", "weak_signal": "放量不涨，说明机构资金仍偏谨慎"}, {"symbol": "sh600584", "name": "长电科技", "role": "封测龙头", "strong_signal": "站上昨收并向设备材料扩散，说明半导体修复有层次", "weak_signal": "弱于AI硬件，说明半导体链未跟上"}, {"symbol": "sz000988", "name": "华工科技", "role": "光模块激光平台", "strong_signal": "与新易盛共振上行，说明光通信修复范围扩大", "weak_signal": "明显弱于龙头，说明中军跟随性不足"}, {"symbol": "sz300502", "name": "新易盛", "role": "CPO总龙头", "strong_signal": "9:45前稳在昨收上方并接近昨高，说明科技修复最真", "weak_signal": "跌回昨收下方并逼近昨低，说明CPO主线承接不足"} ], "cross_cycle_ai_hardware": [ {"symbol": "sz002463", "name": "沪电股份", "role": "高速PCB"}, {"symbol": "sz002475", "name": "立讯精密", "role": "精密制造"}, {"symbol": "sz002241", "name": "歌尔股份", "role": "声学/VR硬件"}, {"symbol": "sz002600", "name": "领益智造", "role": "精密结构件"}, {"symbol": "sz300433", "name": "蓝思科技", "role": "消费电子外观件"}, {"symbol": "sz300735", "name": "光弘科技", "role": "EMS电子制造"}, {"symbol": "sz000021", "name": "深科技", "role": "电子制造"}, {"symbol": "sz000977", "name": "浪潮信息", "role": "AI服务器"}, {"symbol": "sh603019", "name": "中科曙光", "role": "服务器平台"}, {"symbol": "sz300476", "name": "胜宏科技", "role": "高多层PCB"}, {"symbol": "sz002897", "name": "意华股份", "role": "高速连接器"}, {"symbol": "sh601138", "name": "工业富联", "role": "AI服务器制造"}, {"symbol": "sz002837", "name": "英维克", "role": "温控散热"}, {"symbol": "sz002281", "name": "光迅科技", "role": "光器件"}, {"symbol": "sz000988", "name": "华工科技", "role": "光模块/激光"}, {"symbol": "sz300502", "name": "新易盛", "role": "CPO/光模块"}, {"symbol": "sz300620", "name": "光库科技", "role": "光芯片器件"}, {"symbol": "sz300814", "name": "中富电路", "role": "PCB"} ], "cross_cycle_semis": [ {"symbol": "sh688385", "name": "复旦微电", "role": "特种芯片/FPGA"}, {"symbol": "sh688525", "name": "佰维存储", "role": "存储模组"}, {"symbol": "sz300604", "name": "长川科技", "role": "测试设备"}, {"symbol": "sh600171", "name": "上海贝岭", "role": "模拟芯片"}, {"symbol": "sh688041", "name": "海光信息", "role": "国产算力芯片"}, {"symbol": "sz002156", "name": "通富微电", "role": "封测"}, {"symbol": "sz300623", "name": "捷捷微电", "role": "功率半导体"}, {"symbol": "sz300475", "name": "香农芯创", "role": "存储分销"}, {"symbol": "sh600206", "name": "有研新材", "role": "半导体材料"}, {"symbol": "sh600584", "name": "长电科技", "role": "封测龙头"}, {"symbol": "sh603005", "name": "晶方科技", "role": "CIS封装"}, {"symbol": "sz300666", "name": "江丰电子", "role": "靶材"}, {"symbol": "sz000938", "name": "紫光股份", "role": "ICT平台"} ], "cross_cycle_software_platforms": [ {"symbol": "sz300226", "name": "上海钢联", "role": "数据平台"}, {"symbol": "sh600895", "name": "张江高科", "role": "科创平台"}, {"symbol": "sh603383", "name": "顶点软件", "role": "金融IT"}, {"symbol": "sz301171", "name": "易点天下", "role": "出海营销"}, {"symbol": "sz300339", "name": "润和软件", "role": "鸿蒙/AI软件"}, {"symbol": "sz002230", "name": "科大讯飞", "role": "AI应用"}, {"symbol": "sh600536", "name": "中国软件", "role": "信创/操作系统"}, {"symbol": "sz300418", "name": "昆仑万维", "role": "AI平台"} ], "cross_cycle_defense_industrial": [ {"symbol": "sz300775", "name": "三角防务", "role": "航空锻件"}, {"symbol": "sz002977", "name": "天箭科技", "role": "导弹电子"}, {"symbol": "sz002625", "name": "光启技术", "role": "超材料"}, {"symbol": "sz300722", "name": "新余国科", "role": "军工火工"}, {"symbol": "sz002179", "name": "中航光电", "role": "军工连接器"}, {"symbol": "sh600363", "name": "联创光电", "role": "军工光电"}, {"symbol": "sz000099", "name": "中信海直", "role": "低空/通航"}, {"symbol": "sh603728", "name": "鸣志电器", "role": "控制电机"}, {"symbol": "sz300124", "name": "汇川技术", "role": "工控自动化"}, {"symbol": "sz000063", "name": "中兴通讯", "role": "通信设备"} ], "cross_cycle_core": [ {"symbol": "sz002463", "name": "沪电股份", "role": "高速PCB"}, {"symbol": "sz002475", "name": "立讯精密", "role": "精密制造"}, {"symbol": "sh688385", "name": "复旦微电", "role": "特种芯片/FPGA"}, {"symbol": "sz300775", "name": "三角防务", "role": "航空锻件"}, {"symbol": "sz300226", "name": "上海钢联", "role": "数据平台"}, {"symbol": "sh600895", "name": "张江高科", "role": "科创平台"}, {"symbol": "sh603383", "name": "顶点软件", "role": "金融IT"}, {"symbol": "sz002977", "name": "天箭科技", "role": "导弹电子"}, {"symbol": "sh688525", "name": "佰维存储", "role": "存储模组"}, {"symbol": "sz300124", "name": "汇川技术", "role": "工控自动化"}, {"symbol": "sz301171", "name": "易点天下", "role": "出海营销"}, {"symbol": "sz300339", "name": "润和软件", "role": "鸿蒙/AI软件"}, {"symbol": "sz002625", "name": "光启技术", "role": "超材料"}, {"symbol": "sz002241", "name": "歌尔股份", "role": "声学/VR硬件"}, {"symbol": "sz300722", "name": "新余国科", "role": "军工火工"}, {"symbol": "sz300604", "name": "长川科技", "role": "测试设备"}, {"symbol": "sz000063", "name": "中兴通讯", "role": "通信设备"}, {"symbol": "sz002230", "name": "科大讯飞", "role": "AI应用"}, {"symbol": "sz002600", "name": "领益智造", "role": "精密结构件"}, {"symbol": "sh600536", "name": "中国软件", "role": "信创/操作系统"}, {"symbol": "sz300433", "name": "蓝思科技", "role": "消费电子外观件"}, {"symbol": "sz002179", "name": "中航光电", "role": "军工连接器"}, {"symbol": "sz300735", "name": "光弘科技", "role": "EMS电子制造"}, {"symbol": "sh600363", "name": "联创光电", "role": "军工光电"}, {"symbol": "sz000021", "name": "深科技", "role": "电子制造"}, {"symbol": "sz000977", "name": "浪潮信息", "role": "AI服务器"}, {"symbol": "sh600171", "name": "上海贝岭", "role": "模拟芯片"}, {"symbol": "sh688041", "name": "海光信息", "role": "国产算力"}, {"symbol": "sz000099", "name": "中信海直", "role": "低空/通航"}, {"symbol": "sh603728", "name": "鸣志电器", "role": "控制电机"}, {"symbol": "sh603019", "name": "中科曙光", "role": "服务器平台"}, {"symbol": "sz300476", "name": "胜宏科技", "role": "高多层PCB"}, {"symbol": "sz002156", "name": "通富微电", "role": "封测"}, {"symbol": "sz002897", "name": "意华股份", "role": "高速连接器"}, {"symbol": "sh601138", "name": "工业富联", "role": "AI服务器制造"}, {"symbol": "sz002837", "name": "英维克", "role": "温控散热"}, {"symbol": "sz000938", "name": "紫光股份", "role": "ICT平台"}, {"symbol": "sz300623", "name": "捷捷微电", "role": "功率半导体"}, {"symbol": "sz300418", "name": "昆仑万维", "role": "AI平台"}, {"symbol": "sz300475", "name": "香农芯创", "role": "存储分销"}, {"symbol": "sh600206", "name": "有研新材", "role": "半导体材料"}, {"symbol": "sh600584", "name": "长电科技", "role": "封测龙头"}, {"symbol": "sh603005", "name": "晶方科技", "role": "CIS封装"}, {"symbol": "sz300666", "name": "江丰电子", "role": "靶材"}, {"symbol": "sz002281", "name": "光迅科技", "role": "光器件"}, {"symbol": "sz000988", "name": "华工科技", "role": "光模块/激光"}, {"symbol": "sz300502", "name": "新易盛", "role": "CPO/光模块"}, {"symbol": "sz300620", "name": "光库科技", "role": "光芯片器件"}, {"symbol": "sz300814", "name": "中富电路", "role": "PCB"} ], "war_benefit_oil_coal": [ {"symbol": "sh600938", "name": "中国海油", "role": "上游原油", "strong_signal": "继续领涨并扩散到煤炭，说明能源稀缺逻辑仍在强化", "weak_signal": "冲高回落且煤炭跟不上，说明战争溢价开始回吐"}, {"symbol": "sh601857", "name": "中国石油", "role": "综合油气", "strong_signal": "与中国海油共振走强，说明油气链受益扩散", "weak_signal": "弱于中国海油，说明资金只抱最纯上游"}, {"symbol": "sh600028", "name": "中国石化", "role": "综合炼化", "strong_signal": "同步转强，说明资金开始接受综合油化央企", "weak_signal": "明显弱于上游，说明市场只认涨价弹性"}, {"symbol": "sh601088", "name": "中国神华", "role": "煤炭龙头", "strong_signal": "站上昨收并带动煤炭板块，说明避险扩散到煤链", "weak_signal": "强度不及油气，说明煤炭只是在跟随"}, {"symbol": "sh601225", "name": "陕西煤业", "role": "动力煤核心", "strong_signal": "稳中走强，说明高分红煤炭被资金偏好", "weak_signal": "高股息优势失效，说明市场不再追煤炭防御"}, {"symbol": "sh600188", "name": "兖矿能源", "role": "煤炭弹性", "strong_signal": "弹性强于神华，说明资金开始追煤炭进攻属性", "weak_signal": "弹性票不跟，说明煤炭只是红利避险"}, {"symbol": "sh601898", "name": "中煤能源", "role": "煤炭央企", "strong_signal": "跟随神华走强，说明煤炭扩散性良好", "weak_signal": "弱于龙头，说明煤炭内部扩散不足"}, {"symbol": "sh601699", "name": "潞安环能", "role": "区域煤企", "strong_signal": "区域煤企跟涨，说明煤炭板块活跃度上升", "weak_signal": "只剩龙头涨，说明板块修复不完整"}, {"symbol": "sh600256", "name": "广汇能源", "role": "油气煤联动", "strong_signal": "同步走强，说明资金在交易泛能源链", "weak_signal": "弱于原油股，说明市场聚焦更纯粹受益方向"}, {"symbol": "sh600546", "name": "山煤国际", "role": "煤炭贸易", "strong_signal": "贸易股走强，说明煤炭弹性外溢", "weak_signal": "只涨资源不涨贸易，说明市场偏保守"}, {"symbol": "sh601918", "name": "新集能源", "role": "煤电一体", "strong_signal": "煤电一体走强，说明市场在做更宽泛的能源逻辑", "weak_signal": "走弱说明电力属性拖累估值"} ], "war_shock_core12": [ {"symbol": "sh600938", "name": "中国海油", "role": "原油总龙头", "strong_signal": "继续领涨并带动能源链，说明战争溢价仍在抬升", "weak_signal": "冲高回落且能源扩散不足，说明油价冲击开始钝化"}, {"symbol": "sh601857", "name": "中国石油", "role": "综合油气中军", "strong_signal": "与中国海油共振走强，说明油气受益更全面", "weak_signal": "显著弱于海油，说明资金只抱最纯上游"}, {"symbol": "sh601088", "name": "中国神华", "role": "煤炭防御龙头", "strong_signal": "稳步走强，说明避险扩散到煤炭和红利", "weak_signal": "强度不足，说明煤炭只是跟随而非核心"}, {"symbol": "sh601225", "name": "陕西煤业", "role": "动力煤弹性核心", "strong_signal": "强于神华，说明煤炭开始从防御转向进攻", "weak_signal": "弱于神华，说明资金只偏好稳健红利煤"}, {"symbol": "sh600256", "name": "广汇能源", "role": "泛能源弹性票", "strong_signal": "跟随油煤同步放大，说明市场在交易更宽泛能源短缺", "weak_signal": "掉队说明资金只做央企大票"}, {"symbol": "sz000977", "name": "浪潮信息", "role": "算力风向标", "strong_signal": "若逆势翻红，说明科技开始消化战争冲击", "weak_signal": "持续水下说明战争环境仍压制算力风格"}, {"symbol": "sh688041", "name": "海光信息", "role": "高弹性国产算力", "strong_signal": "快速止跌翻红，说明高弹性风险偏好修复", "weak_signal": "继续走弱说明成长定价仍被压缩"}, {"symbol": "sh603019", "name": "中科曙光", "role": "服务器平台", "strong_signal": "跟随浪潮和海光企稳，说明平台链承接变好", "weak_signal": "平台持续弱势说明算力产业链仍受压"}, {"symbol": "sz300502", "name": "新易盛", "role": "CPO龙头", "strong_signal": "逆势转强意味着战争扰动正在被科技主线吸收", "weak_signal": "龙头失守说明最强科技也扛不住宏观冲击"}, {"symbol": "sh603881", "name": "数据港", "role": "IDC运营", "strong_signal": "企稳说明市场未继续惩罚高电耗IDC", "weak_signal": "持续下跌说明电力成本与风险偏好双重压制仍在"}, {"symbol": "sh600011", "name": "华能国际", "role": "火电成本代表", "strong_signal": "若火电能稳住，说明成本冲击预期可控", "weak_signal": "走弱说明燃料涨价正在压制火电估值"}, {"symbol": "sh600023", "name": "浙能电力", "role": "区域火电对照", "strong_signal": "区域火电跟稳，说明电力防御属性仍有效", "weak_signal": "同步转弱说明传统火电整体受成本约束"} ], "war_headwind_compute_power": [ {"symbol": "sz000977", "name": "浪潮信息", "role": "AI服务器", "strong_signal": "若逆势翻红，说明算力链正在消化战争冲击", "weak_signal": "弱于油煤且始终水下，说明战争环境压制算力风险偏好"}, {"symbol": "sh603019", "name": "中科曙光", "role": "服务器平台", "strong_signal": "跟随浪潮信息企稳，说明平台链抗压", "weak_signal": "走弱说明平台层最先承压"}, {"symbol": "sh601138", "name": "工业富联", "role": "AI服务器制造", "strong_signal": "大票制造抗跌，说明机构并未全面撤离算力链", "weak_signal": "放量走弱，说明机构在降配高耗能硬件链"}, {"symbol": "sh688041", "name": "海光信息", "role": "国产算力芯片", "strong_signal": "高弹性核心止跌翻红，说明风险偏好有回流", "weak_signal": "高弹性继续走弱，说明成长定价被压缩"}, {"symbol": "sz300308", "name": "中际旭创", "role": "光模块", "strong_signal": "若光模块重新领涨，说明市场重回科技主线", "weak_signal": "明显弱于油煤，说明风险偏好未恢复"}, {"symbol": "sz300502", "name": "新易盛", "role": "CPO龙头", "strong_signal": "逆势转强可视作战争扰动减弱", "weak_signal": "龙头失守说明高景气科技也扛不住宏观冲击"}, {"symbol": "sz002837", "name": "英维克", "role": "温控散热", "strong_signal": "抗跌说明算力配套链仍有配置需求", "weak_signal": "弱于主链说明配套端先被抛售"}, {"symbol": "sh600845", "name": "宝信软件", "role": "IDC平台", "strong_signal": "稳于IDC同行，说明数据中心平台更抗压", "weak_signal": "走弱说明高电耗预期压制IDC估值"}, {"symbol": "sh603881", "name": "数据港", "role": "IDC运营", "strong_signal": "企稳说明市场未继续惩罚高电耗模型", "weak_signal": "下跌说明战争引发的电价与风险偏好双杀仍在"}, {"symbol": "sz300738", "name": "奥飞数据", "role": "IDC弹性", "strong_signal": "弹性股翻红说明情绪开始回暖", "weak_signal": "弹性持续最弱说明资金仍回避IDC"}, {"symbol": "sh600011", "name": "华能国际", "role": "火电", "strong_signal": "若火电抗跌，说明市场认为成本冲击可控", "weak_signal": "走弱说明燃料成本上升压制火电盈利"}, {"symbol": "sh600027", "name": "华电国际", "role": "火电", "strong_signal": "跟随稳住，说明火电链并未被系统性抛售", "weak_signal": "明显走弱说明成本端担忧升温"}, {"symbol": "sh600795", "name": "国电电力", "role": "综合电力", "strong_signal": "企稳说明市场更看重央企防守属性", "weak_signal": "回落说明电力板块也受成本预期拖累"}, {"symbol": "sh600023", "name": "浙能电力", "role": "区域火电", "strong_signal": "稳住说明区域火电防御性尚可", "weak_signal": "走弱说明资金回避传统火电"} ] } FILE:assets/industry_chains.json { "themes": [ { "id": "optical_module_chain", "label": "光模块产业链", "query": "A股光模块产业链股票", "categories": ["huge_future", "huge_name_release"], "triggers": ["光模块", "cpo", "光通信", "光芯片", "铜缆", "光器件", "datacenter", "data center"], "preferred_groups": ["tech_repair", "cross_cycle_ai_hardware", "cross_cycle_anchor12"], "strong_signal": "产业链核心票同步转强，说明最强科技主线正在扩散。", "weak_signal": "只有单一龙头强，链条中军和二线不跟，说明产业链扩散不足。" }, { "id": "compute_power_chain", "label": "算力产业链", "query": "A股算力产业链股票", "categories": ["huge_future", "huge_name_release"], "triggers": ["算力", "服务器", "液冷", "数据中心", "idc", "server", "compute", "cooling"], "preferred_groups": ["tech_repair", "cross_cycle_ai_hardware", "war_headwind_compute_power"], "strong_signal": "服务器、液冷、IDC 同步改善，说明算力链从点状修复转向面状修复。", "weak_signal": "只有少数硬件股强，平台链和IDC仍弱，说明算力修复不完整。" }, { "id": "semiconductor_chain", "label": "半导体产业链", "query": "A股半导体产业链股票", "categories": ["huge_future", "huge_name_release"], "triggers": ["芯片", "半导体", "gpu", "gpus", "存储", "semiconductor", "chip", "chips"], "preferred_groups": ["cross_cycle_semis", "cross_cycle_anchor12", "tech_repair"], "strong_signal": "芯片、设备、封测同步企稳，说明科技修复有深度。", "weak_signal": "高弹性芯片继续被压制，说明半导体只是跟随反弹。" }, { "id": "robotics_chain", "label": "机器人与低空链", "query": "A股机器人产业链股票", "categories": ["huge_future", "huge_name_release"], "triggers": ["机器人", "humanoid", "robot", "drone", "无人机", "商业航天", "低空"], "preferred_groups": ["cross_cycle_defense_industrial", "cross_cycle_anchor12"], "strong_signal": "机器人与控制零部件同步走强，说明未来叙事开始向制造端扩散。", "weak_signal": "题材热但中军不跟，说明仍停留在消息脉冲。" }, { "id": "oil_gas_chain", "label": "油气受益链", "query": "A股石油天然气概念股", "categories": ["huge_conflict"], "triggers": ["oil", "crude", "brent", "wti", "lng", "gas", "shipping", "hormuz", "gulf", "iran", "israel"], "preferred_groups": ["war_benefit_oil_coal", "war_shock_core12", "defensive_gauge"], "strong_signal": "上游油气和综合能源央企共振走强，说明战争溢价仍在强化。", "weak_signal": "只有最纯上游强，综合能源和煤炭不跟，说明受益扩散不足。" }, { "id": "coal_chain", "label": "煤炭防御链", "query": "A股煤炭概念股", "categories": ["huge_conflict"], "triggers": ["coal", "energy", "动力煤", "煤炭", "神华", "coal mine"], "preferred_groups": ["war_benefit_oil_coal", "defensive_gauge", "war_shock_core12"], "strong_signal": "神华与弹性煤企一起走强，说明防御抱团正在向煤链扩散。", "weak_signal": "只有红利煤强，弹性煤企不跟，说明只是稳健避险。" }, { "id": "idc_power_chain", "label": "IDC与电力成本链", "query": "A股数据中心概念股", "categories": ["huge_conflict", "huge_future"], "triggers": ["electricity", "power", "data center", "datacenter", "idc", "电力", "电价"], "preferred_groups": ["war_headwind_compute_power", "cross_cycle_ai_hardware"], "strong_signal": "IDC 与温控链稳住，说明市场未继续惩罚高电耗资产。", "weak_signal": "IDC 和火电同步走弱，说明成本冲击仍在压制估值。" } ], "group_theme_hints": { "core10": ["optical_module_chain", "compute_power_chain", "oil_gas_chain"], "tech_repair": ["optical_module_chain", "compute_power_chain", "semiconductor_chain"], "policy_beta": ["compute_power_chain"], "defensive_gauge": ["oil_gas_chain", "coal_chain"], "cross_cycle_anchor12": ["optical_module_chain", "compute_power_chain", "semiconductor_chain"], "cross_cycle_ai_hardware": ["optical_module_chain", "compute_power_chain"], "cross_cycle_semis": ["semiconductor_chain"], "cross_cycle_software_platforms": ["compute_power_chain"], "cross_cycle_defense_industrial": ["robotics_chain"], "war_benefit_oil_coal": ["oil_gas_chain", "coal_chain"], "war_headwind_compute_power": ["compute_power_chain", "idc_power_chain"], "war_shock_core12": ["oil_gas_chain", "coal_chain", "idc_power_chain"] } } FILE:assets/mx_presets.json { "preopen_policy": { "description": "Step 1 policy and official-news scan before the A-share open.", "steps": [ { "slug": "policy_news", "tool": "news-search", "query": "今日国务院发改委工信部证监会最新政策 A股影响", "size": 8, "limit": 5 } ] }, "preopen_global_risk": { "description": "Step 1 overnight US, oil, and commodity risk scan.", "steps": [ { "slug": "us_equities", "tool": "news-search", "query": "隔夜美股纳指英伟达特斯拉苹果最新资讯 A股影响", "size": 8, "limit": 5 }, { "slug": "oil_commodities", "tool": "news-search", "query": "布伦特原油铜黄金煤炭最新资讯 A股影响", "size": 8, "limit": 5 } ] }, "board_optical_module": { "description": "Step 2 board resonance scan for optical module and CPO names.", "steps": [ { "slug": "optical_module_screen", "tool": "stock-screen", "keyword": "A股光模块概念股", "page_no": 1, "page_size": 12, "limit": 10 } ] }, "board_compute_power": { "description": "Step 2 board resonance scan for compute-power names.", "steps": [ { "slug": "compute_power_screen", "tool": "stock-screen", "keyword": "A股算力概念股", "page_no": 1, "page_size": 12, "limit": 10 } ] }, "board_energy_defense": { "description": "Temporary overlay scan for energy and defensive beneficiaries.", "steps": [ { "slug": "energy_defense_screen", "tool": "stock-screen", "keyword": "A股石油煤炭板块股票", "page_no": 1, "page_size": 12, "limit": 10 } ] }, "flow_main_force": { "description": "Capital-flow check for the whole market plus top inflow and outflow names.", "steps": [ { "slug": "market_main_flow", "tool": "query", "tool_query": "全部A股主力净流入资金今日大单净流入中单净流入小单净流入", "limit": 10 }, { "slug": "top_main_force_inflow", "tool": "stock-screen", "keyword": "A股主力资金净流入前20股票", "page_no": 1, "page_size": 20, "limit": 10 }, { "slug": "top_main_force_outflow", "tool": "stock-screen", "keyword": "A股主力资金净流出前20股票", "page_no": 1, "page_size": 20, "limit": 10 } ] }, "chain_optical_module": { "description": "Industry-chain scan for optical module and CPO names.", "steps": [ { "slug": "optical_module_chain", "tool": "stock-screen", "keyword": "A股光模块产业链股票", "page_no": 1, "page_size": 15, "limit": 10 } ] }, "chain_conflict_energy": { "description": "Industry-chain overlay for conflict-beneficiary energy names.", "steps": [ { "slug": "oil_gas_chain", "tool": "stock-screen", "keyword": "A股石油天然气概念股", "page_no": 1, "page_size": 15, "limit": 10 }, { "slug": "coal_chain", "tool": "stock-screen", "keyword": "A股煤炭概念股", "page_no": 1, "page_size": 15, "limit": 10 } ] }, "validate_inspur": { "description": "Structured data check for Inspur core trading metrics.", "steps": [ { "slug": "inspur_metrics", "tool": "query", "tool_query": "浪潮信息最新价总市值收盘价", "limit": 8 } ] }, "validate_luxshare": { "description": "Structured data check for Luxshare and AI interconnect cues.", "steps": [ { "slug": "luxshare_metrics", "tool": "query", "tool_query": "立讯精密最新价总市值收盘价", "limit": 8 } ] }, "preopen_repair_chain": { "description": "A compact desk workflow: overnight risk, CPO board screen, and Inspur validation.", "steps": [ { "slug": "us_equities", "tool": "news-search", "query": "隔夜美股纳指英伟达特斯拉苹果最新资讯 A股影响", "size": 8, "limit": 5 }, { "slug": "optical_module_screen", "tool": "stock-screen", "keyword": "A股光模块概念股", "page_no": 1, "page_size": 12, "limit": 10 }, { "slug": "inspur_metrics", "tool": "query", "tool_query": "浪潮信息最新价总市值收盘价", "limit": 8 } ] } } FILE:assets/news_iterator_config.json { "feeds": [ { "key": "google-top", "label": "Google News Top Stories", "url": "https://news.google.com/rss?hl=en-US&gl=US&ceid=US:en" }, { "key": "google-business", "label": "Google News Business Search", "url": "https://news.google.com/rss/search?q=business+markets+technology&hl=en-US&gl=US&ceid=US:en" }, { "key": "google-ai", "label": "Google News AI Search", "url": "https://news.google.com/rss/search?q=AI+model+launch+chip+datacenter+robot&hl=en-US&gl=US&ceid=US:en" }, { "key": "google-big-names", "label": "Google News Big Names Search", "url": "https://news.google.com/rss/search?q=OpenAI+OR+NVIDIA+OR+Microsoft+OR+Meta+OR+Google+OR+Apple+OR+Tesla+OR+xAI&hl=en-US&gl=US&ceid=US:en" }, { "key": "google-conflict", "label": "Google News Conflict Search", "url": "https://news.google.com/rss/search?q=Middle+East+oil+conflict+Iran+Israel+shipping+energy&hl=en-US&gl=US&ceid=US:en" }, { "key": "bbc-world", "label": "BBC World", "url": "https://feeds.bbci.co.uk/news/world/rss.xml" }, { "key": "nyt-world", "label": "NYT World", "url": "https://rss.nytimes.com/services/xml/rss/nyt/World.xml" }, { "key": "wsj-world", "label": "WSJ World", "url": "https://feeds.a.dj.com/rss/RSSWorldNews.xml" } ], "big_name_entities": [ "OpenAI", "NVIDIA", "Jensen Huang", "Sam Altman", "Microsoft", "Meta", "Mark Zuckerberg", "Google", "Alphabet", "Sundar Pichai", "Apple", "Tim Cook", "Tesla", "Elon Musk", "xAI", "Amazon", "Anthropic", "TSMC", "ASML", "AMD", "Intel", "Broadcom", "Oracle" ], "conflict_entities": [ "Iran", "Israel", "Middle East", "Hormuz", "Gulf", "Hezbollah", "Houthis", "Red Sea", "OPEC" ], "future_keywords": [ "breakthrough", "next-generation", "new model", "models", "foundation model", "agent", "reasoning", "inference", "chip", "chips", "gpu", "gpus", "semiconductor", "semiconductors", "datacenter", "data center", "robot", "robots", "humanoid", "autonomous", "battery", "batteries", "fusion", "quantum", "satellite", "satellites", "drone", "drones" ], "release_verbs": [ "launch", "launched", "launches", "release", "released", "releases", "announce", "announced", "announces", "unveil", "unveiled", "unveils", "introduce", "introduced", "introduces", "ship", "shipping", "ships", "debut", "debuts" ], "conflict_keywords": [ "war", "conflict", "attack", "attacks", "airstrike", "airstrikes", "missile", "missiles", "drone strike", "drone strikes", "retaliation", "sanction", "sanctions", "blockade", "shipping disruption", "hormuz", "gulf", "oil facility", "gas field", "refinery" ], "energy_keywords": [ "oil", "crude", "brent", "wti", "lng", "gas", "coal", "energy", "refinery", "shipping" ], "compute_power_keywords": [ "ai server", "ai servers", "server", "servers", "datacenter", "datacenters", "data center", "data centers", "compute", "power price", "power grid", "electricity", "idc", "cooling", "thermal" ], "keyword_watchlists": { "ai": ["cross_cycle_software_platforms", "cross_cycle_anchor12"], "model": ["cross_cycle_software_platforms", "cross_cycle_anchor12"], "agent": ["cross_cycle_software_platforms", "cross_cycle_anchor12"], "chip": ["cross_cycle_semis", "cross_cycle_anchor12"], "gpu": ["cross_cycle_semis", "cross_cycle_anchor12"], "semiconductor": ["cross_cycle_semis", "cross_cycle_anchor12"], "datacenter": ["cross_cycle_ai_hardware", "cross_cycle_anchor12"], "data center": ["cross_cycle_ai_hardware", "cross_cycle_anchor12"], "server": ["cross_cycle_ai_hardware", "cross_cycle_anchor12"], "robot": ["cross_cycle_defense_industrial", "cross_cycle_anchor12"], "humanoid": ["cross_cycle_defense_industrial", "cross_cycle_anchor12"], "drone": ["cross_cycle_defense_industrial", "cross_cycle_anchor12"], "oil": ["war_benefit_oil_coal", "war_shock_core12", "defensive_gauge"], "crude": ["war_benefit_oil_coal", "war_shock_core12", "defensive_gauge"], "coal": ["war_benefit_oil_coal", "war_shock_core12", "defensive_gauge"], "energy": ["war_benefit_oil_coal", "war_shock_core12", "defensive_gauge"], "compute": ["war_headwind_compute_power", "war_shock_core12"], "idc": ["war_headwind_compute_power", "war_shock_core12"], "electricity": ["war_headwind_compute_power", "war_shock_core12"], "power": ["war_headwind_compute_power", "war_shock_core12"] }, "entity_watchlists": { "openai": ["cross_cycle_software_platforms", "cross_cycle_anchor12"], "nvidia": ["cross_cycle_ai_hardware", "cross_cycle_semis", "cross_cycle_anchor12"], "jensen huang": ["cross_cycle_ai_hardware", "cross_cycle_semis", "cross_cycle_anchor12"], "sam altman": ["cross_cycle_software_platforms", "cross_cycle_anchor12"], "microsoft": ["cross_cycle_software_platforms", "cross_cycle_ai_hardware", "cross_cycle_anchor12"], "meta": ["cross_cycle_software_platforms", "cross_cycle_ai_hardware", "cross_cycle_anchor12"], "mark zuckerberg": ["cross_cycle_software_platforms", "cross_cycle_anchor12"], "google": ["cross_cycle_software_platforms", "cross_cycle_anchor12"], "alphabet": ["cross_cycle_software_platforms", "cross_cycle_anchor12"], "apple": ["cross_cycle_ai_hardware", "cross_cycle_anchor12"], "tesla": ["cross_cycle_ai_hardware", "cross_cycle_defense_industrial", "cross_cycle_anchor12"], "elon musk": ["cross_cycle_ai_hardware", "cross_cycle_defense_industrial", "cross_cycle_anchor12"], "xai": ["cross_cycle_software_platforms", "cross_cycle_anchor12"], "tsmc": ["cross_cycle_semis", "cross_cycle_anchor12"], "asml": ["cross_cycle_semis", "cross_cycle_anchor12"], "amd": ["cross_cycle_semis", "cross_cycle_anchor12"], "intel": ["cross_cycle_semis", "cross_cycle_anchor12"], "broadcom": ["cross_cycle_semis", "cross_cycle_ai_hardware", "cross_cycle_anchor12"], "oracle": ["cross_cycle_ai_hardware", "cross_cycle_software_platforms", "cross_cycle_anchor12"], "iran": ["war_benefit_oil_coal", "war_headwind_compute_power", "war_shock_core12", "defensive_gauge"], "israel": ["war_benefit_oil_coal", "war_headwind_compute_power", "war_shock_core12", "defensive_gauge"], "middle east": ["war_benefit_oil_coal", "war_headwind_compute_power", "war_shock_core12", "defensive_gauge"], "hormuz": ["war_benefit_oil_coal", "war_headwind_compute_power", "war_shock_core12", "defensive_gauge"], "gulf": ["war_benefit_oil_coal", "war_headwind_compute_power", "war_shock_core12", "defensive_gauge"] } } FILE:references/cross-cycle-watchlist.md # Cross-Cycle Watchlist This skill now includes a larger cross-cycle A-share quality pool built from names that tend to survive multiple bull-bear cycles and repeatedly re-emerge as leaders when a new risk-on phase starts. ## The Problem A good watchlist becomes useless if it is too large to act on. Do not treat the full pool as a buy list. Treat it as a priority universe. ## Built-In Groups - `cross_cycle_anchor12` - Daily-use anchor list. Start here every morning. - `cross_cycle_ai_hardware` - PCB, optical, AI server, manufacturing, thermal, and hardware supply-chain names. - `cross_cycle_semis` - Chips, packaging, testing, storage, and materials. - `cross_cycle_software_platforms` - AI software, data/platform, and domestic software names. - `cross_cycle_defense_industrial` - Defense, industrial automation, and related advanced manufacturing. - `cross_cycle_core` - Full pool. Best used for weekly review or major phase-rotation scans. ## How To Use The Pool ### 1. Daily Pre-Open Use: - `cross_cycle_anchor12` - one or two theme buckets relevant to the current tape Goal: - reduce the market to a small set of high-quality leaders before the open ### 2. First 30 Minutes If the market is trying to repair: - check which names from `cross_cycle_anchor12` reclaim prior close first - prefer the names that are both liquid and early - do not prioritize the biggest loser in the full pool ### 3. Pullback Days Use the pool to find resilience, not cheapness. Look for: - names down less than their theme peers - names that do not lose the prior swing low - names that keep turnover and institutional sponsorship ### 4. Rotation Days Use the theme groups to identify where the market is paying up: - `cross_cycle_ai_hardware`: hardware, optics, PCB, AI infrastructure - `cross_cycle_semis`: chips, packaging, testing, storage, materials - `cross_cycle_software_platforms`: software and application-side risk appetite - `cross_cycle_defense_industrial`: defense and hard-tech manufacturing rotation ## Practical Workflow 1. Start with `cross_cycle_anchor12`. 2. If leadership is obvious, zoom into the matching theme bucket. 3. If the market is broad and strong, widen to `cross_cycle_core`. 4. If the tape is weak, narrow back down to the earliest reclaimers only. ## What Not To Do - Do not watch all names equally every day. - Do not bottom-fish the weakest subgroup just because it fell the most. - Do not treat every name here as a simultaneous candidate. - Do not confuse long-term quality with short-term timing. ## Recommended Defaults - Daily: `cross_cycle_anchor12` - Theme check: one or two matching buckets - Weekly review: `cross_cycle_core` FILE:references/data-sources.md # Data Sources Use primary and official sources first when the conclusion depends on policy, rates, or geopolitics. ## Official And Primary Sources | Source | Use | Notes | |---|---|---| | `pbc.gov.cn` | `LPR`, PBOC guidance, domestic policy timing | Use exact release dates and times | | `federalreserve.gov` | FOMC statements and projections | Use official statement pages for exact wording | | `bankofengland.co.uk` | BOE policy decisions | Useful when global rates affect risk appetite | | `snb.ch` | SNB policy decisions | Secondary global rates input | | `apnews.com` / `reuters.com` | Geopolitics, oil, fast macro news | Use for timely cross-market context | ## Market Data Endpoints In This Skill ## Source Priority | Layer | Primary | Secondary | Benchmark Command | |---|---|---|---| | Index and sector tape | Eastmoney public endpoints | none in this skill | `python3 scripts/benchmark_sources.py --skip-mx` | | Liquid stock quotes | Tencent quote snapshot | none in this skill | `python3 scripts/benchmark_sources.py --skip-mx` | | High-attention event intake | Google News / BBC / NYT / WSJ RSS | direct web verification | `python3 scripts/benchmark_sources.py --skip-mx` | | Financial news search | `mx_toolkit.py news-search` | public RSS + primary websites | `python3 scripts/benchmark_sources.py` | | Stock screening | `mx_toolkit.py stock-screen` | static watchlists | `python3 scripts/benchmark_sources.py` | | Structured security data | `mx_toolkit.py query` | public quote snapshot for a quick check | `python3 scripts/benchmark_sources.py` | ### Tencent Quote Snapshot - Endpoint: `https://qt.gtimg.cn/q=...` - Use: liquid stock watchlists - Encoding: `GBK` - Key parsed fields: - name - code - price - prior close - open - high - low - absolute change - percent change - amount - timestamp ### Eastmoney Index Snapshot - Endpoint: `https://push2.eastmoney.com/api/qt/ulist.np/get` - Use: main index levels and simple breadth - Default symbols: - `1.000001` Shanghai Composite - `0.399001` Shenzhen Component - `0.399006` ChiNext - `1.000300` CSI 300 - `1.000688` STAR 50 - `0.899050` Beijing 50 ### Eastmoney Sector Breadth - Endpoint: `https://push2.eastmoney.com/api/qt/clist/get` - Use: strongest and weakest sectors on the day - The skill defaults to concept-board ranking with the common Eastmoney parameters already set in the script. ### MX News Search - Endpoint: `https://mkapi2.dfcfs.com/finskillshub/api/claw/news-search` - Use: timely finance news search with stronger source coverage than public RSS - Entry point in this skill: - `python3 scripts/mx_toolkit.py news-search --query '立讯精密最新资讯'` - `python3 scripts/mx_toolkit.py preset --name preopen_policy` ### MX Stock Screen - Endpoint: `https://mkapi2.dfcfs.com/finskillshub/api/claw/stock-screen` - Use: natural-language board and stock screening - Entry point in this skill: - `python3 scripts/mx_toolkit.py stock-screen --keyword 'A股光模块概念股'` - `python3 scripts/mx_toolkit.py preset --name board_optical_module` ### MX Structured Data Query - Endpoint: `https://mkapi2.dfcfs.com/finskillshub/api/claw/query` - Use: entity-level structured data such as latest price, market cap, and time-series tables - Entry point in this skill: - `python3 scripts/mx_toolkit.py query --tool-query '浪潮信息最新价总市值收盘价'` - `python3 scripts/mx_toolkit.py preset --name validate_inspur` ## Practical Notes - These public endpoints may throttle or change. - Use scripts for fast snapshots, not for blindly trusting a single source. - Use `benchmark_sources.py` before assigning a source as the primary feed for a live session. - If the user asks for the latest or today’s view, verify the unstable facts live before drawing conclusions. FILE:references/event-regime-watchlists.md # Event Regime Watchlists Use these lists when the market enters a temporary shock regime driven by war, shipping disruption, or an energy-price spike. ## Principle Do not permanently promote event-regime lists into the core daily framework. They are overlays. Use them only when: - Middle East conflict escalates - oil and LNG spike sharply - shipping or supply routes are disrupted - inflation fears reprice global risk assets ## Built-In Groups - `war_shock_core12` - The smallest practical wartime overlay. Use this first when you only want the most important regime indicators. - `war_benefit_oil_coal` - Oil, gas, coal, and upstream energy names that usually benefit from a sustained energy shock. - `war_headwind_compute_power` - Compute, IDC, and thermal-power names that can face valuation, cost, or risk-appetite pressure during an energy-led shock. ## How To Use Them ### 1. Regime Detection First confirm the market is actually in an energy-shock regime: - crude spikes sharply - coal chain strengthens - defensives outperform growth - compute infrastructure lags despite otherwise good fundamentals ### 2. Relative Observation If time is limited, start with `war_shock_core12`. Use `war_benefit_oil_coal` to see whether the market is paying for energy scarcity. Use `war_headwind_compute_power` to check whether high-power-consumption and thermal-power groups are under pressure. ### 3. Exit Criteria Downgrade these overlays when: - oil stops trending higher - defensive leadership fades - growth reclaims leadership breadth - the geopolitical shock de-escalates ## What Not To Do - Do not confuse a short-term war shock with a long-cycle structural bull thesis. - Do not keep trading war overlays after the shock has faded. - Do not use the headwind list as an automatic short list without confirming price action and breadth. ## Recommended Defaults - first check: `war_shock_core12` - expansion: `war_benefit_oil_coal` and `war_headwind_compute_power` - exit: remove the overlay when oil and defensive leadership stop dominating FILE:references/message-iterator.md # Message Iterator This module is for persistent news intake. It continuously polls public RSS feeds, scores headlines, and stores high-signal alerts into a local SQLite database. It also converts those alerts into event-driven stock pools automatically, so the desk does not wait for manual watchlist updates. ## Target Alert Types 1. `huge_future` - Something with unusually large future potential. - Examples: AI model breakthroughs, new chips, data-center buildouts, robots, batteries, quantum, fusion. 2. `huge_name_release` - A globally famous company or person releases or announces something. - Examples: OpenAI, NVIDIA, Microsoft, Meta, Google, Apple, Tesla, xAI, Jensen Huang, Sam Altman, Elon Musk. 3. `huge_conflict` - A major conflict, strike, sanction, or energy-route disruption. - Examples: Middle East escalation, Hormuz threats, refinery strikes, shipping disruption. ## What The Iterator Produces For each high-signal item it stores: - title - link - source feed - published time - matched categories - matched entities and keywords - impacted watchlist groups - score and signal strength For each reporting window it also builds: - `event_focus_huge_conflict_benefit` - `event_focus_huge_conflict_headwind` - `event_focus_huge_conflict_defensive` - `event_focus_huge_future` - `event_focus_huge_name_release` These pools are written into `event_watchlists.json` and can be pulled directly into the morning brief and opening checklist. ## Default Market Mapping - `huge_future` - `cross_cycle_ai_hardware` - `cross_cycle_semis` - `cross_cycle_software_platforms` - `cross_cycle_anchor12` - `huge_name_release` - mapped by entity, with big-tech releases usually flowing to the same technology groups - `huge_conflict` - `war_shock_core12` - `war_benefit_oil_coal` - `war_headwind_compute_power` - `defensive_gauge` ## Run Modes ### One-Off Poll ```bash python3 scripts/news_iterator.py poll ``` ### Continuous Loop ```bash python3 scripts/news_iterator.py loop --interval-seconds 300 ``` ### Generate A Report ```bash python3 scripts/news_iterator.py report --hours 12 ``` ## Long-Running Deployment The simplest portable deployment is: ```bash nohup python3 scripts/news_iterator.py loop --interval-seconds 300 > ~/uwillberich-news-iterator.log 2>&1 & ``` On macOS, the better deployment is `launchd`. This runs one poll on a fixed interval instead of keeping a Python process alive forever: ```bash python3 scripts/install_news_iterator_launchd.py install --interval-seconds 300 python3 scripts/install_news_iterator_launchd.py status ``` The script stores state under: - `~/.uwillberich/news-iterator/` By default it writes: - `news_iterator.sqlite3` - `latest_alerts.md` - `alerts.jsonl` - `event_watchlists.json` ## Practical Workflow 1. Let the iterator run in the background. 2. Check the markdown report when you prepare the next session. 3. Let the auto-generated event stock pools flow into the desk reports. 4. If the top alerts skew to `huge_conflict`, use the split pools: - benefit: oil and coal - headwind: compute power, IDC, and power names - defensive: low-volatility shelters 5. If the top alerts skew to `huge_future` or `huge_name_release`, narrow into the generated technology pool first, then the static quality pools. ## Classification Notes - Matching is term-boundary aware, so short words like `AI` do not trigger on unrelated text such as `said`. - Conflict entities are tracked separately from big-name entities, so `Iran`, `Israel`, `Hormuz`, and similar terms can trigger the war overlay directly. - The markdown snapshot is a rolling lookback report, while `alerts.jsonl` stays append-only for audit and later replay. ## What Not To Do - Do not treat every alert as actionable. - Do not confuse raw attention with sustained market leadership. - Do not let conflict overlays permanently replace the core quality pools. FILE:references/methodology.md # Methodology ## Objective Convert recent market action and overnight catalysts into a next-session A-share plan. The framework remains decision-first and tape-aware. It absorbs event-driven and breakout logic, but it does not give up the current discipline around policy timing, breadth, and relative strength. ## Governing Question Before asking whether something is "good" or "bad," ask: - Can this event break out beyond a niche trading circle? - Can it attract large capital, not just retail chatter? - Does it map to a recognizable board, supply chain, or second-order beneficiary in A-shares? If the answer is no, downgrade it quickly. ## Attention-To-Liquidity Chain Use this chain as the base mental model: `event attention -> large-capital focus -> board activation -> retail participation -> liquidity expansion -> distribution` This does not mean every attention event becomes a trade. It means the real job is to judge where the event currently sits in the chain and whether it is accelerating or already entering distribution. ## Market State Classifier Classify the tape into one of three states before discussing stock ideas. ### 1. Mainline Market Typical traits: - national-level or global catalyst - policy or geopolitical event with broad public attention - clear board resonance, not just one or two names - several liquid leaders moving together - strong breadth or strong improvement in breadth High-conviction confirmation can include: - many limit-up names in one board - multiple leaders with real turnover and repeated follow-through - brokers or broad risk proxies confirming improving sentiment Default response: - focus on the core board - prefer leaders and strong second-line names with liquidity - allow more aggressive follow-through than in range conditions ### 2. Independent-Leader Market Typical traits: - only one to three names detach from the rest - the board itself is weak or mixed - catalyst is company-specific: orders, earnings, restructuring, asset injection, price shock, or seasonal theme - liquidity concentrates into a few symbols rather than spreading Default response: - trade the specific leader or very narrow chain - do not extrapolate a full board thesis unless follow-through broadens - keep horizon shorter than in a true mainline market ### 3. Range Or Defensive Market Typical traits: - no convincing mainline - no durable independent leader - breadth is poor or inconsistent - strength sits in oil, coal, banks, telecom, utilities, or other low-beta shelters Default response: - shorten holding period - avoid blind breakout chasing - prefer observation, selective mean reversion, or defensive bias ## Core Trading Principles 1. Relative strength beats blind bottom-fishing. - In weak tape, the strongest surviving leaders are usually better repair candidates than the biggest losers. 2. Separate repair from defense. - If strength stays concentrated in oil, coal, banks, telecom, or utilities, the market is not broadly healthy. 3. Breadth confirms, index alone does not. - A green index with weak mid-cap breadth is often low-quality. 4. Policy timing matters. - On `LPR` dates, the `09:00` release can override a purely external-risk setup. 5. Board logic must be understandable. - If retail can understand the causal chain quickly, breakout probability is higher. 6. Prefer second-order beneficiaries over crowded first-order intuition when appropriate. - In geopolitical or policy shocks, the obvious trade is often too crowded. The better trade can sit one logical layer deeper. 7. Match holding horizon to catalyst duration. - A one-day squeeze, a multi-week policy cycle, and a long-duration value repair should not be treated the same way. 8. Distribution matters as much as ignition. - If a big leader stalls while small caps rotate in sequence, the board may be entering exit liquidity rather than new expansion. ## Catalyst Hierarchy Rank event quality roughly in this order: 1. national-level policy or multi-ministry direction 2. global conflict, disaster, or technology breakthrough 3. fixed-date public anchors with high attention 4. sector-specific price shocks or industry data surprises 5. company-specific catalysts 6. pure sentiment or name-based speculation The higher the event sits on this ladder, the more seriously you should consider a mainline scenario. ## Event Templates Worth Keeping ### Strong Policy / National Event - good candidate for mainline classification - especially strong when date, scale, and public attention are explicit ### Commodity Price Shock - good fit for `SHORT -> MID` style thinking - scan upstream A-share beneficiaries immediately after price confirmation ### Geopolitical Shock - first ask what the direct board is - then ask what the cleaner second-order beneficiary is - prefer the second-order logic if it is more differentiated and still easy to understand ### Company-Specific Breakout - usually belongs in independent-leader mode - promote it to a board thesis only if followers appear with volume ### Sentiment-Only Speculation - treat as a thermometer, not a core recommendation - useful for judging heat, dangerous as a default strategy ## Trade Horizon Alignment Use horizon labels only as a thinking aid, not a hardcoded portfolio engine. ### SHORT Use when: - move is steep and abnormal - catalyst is fast-money friendly - single-day or multi-day squeeze behavior dominates Default posture: - faster profit-taking - do not assume durability without a second day of confirmation ### MID Use when: - catalyst can sustain for days or weeks - board resonance or trend continuation exists - liquidity stays healthy without one-day blowoff behavior Default posture: - favor trend follow-through and pullback entries - keep validating with breadth and leader behavior ### LONG Keep as a separate mindset only when: - the user explicitly asks for slower fundamental accumulation - the thesis is valuation or quality driven rather than next-session tactical This skill is not primarily a long-horizon portfolio system, so LONG ideas should remain secondary. ## Three-Layer Framework ### 1. External Shock Layer Check: - oil and gas price shocks - geopolitics - U.S. rates and central-bank messaging - U.S. and Hong Kong index direction Interpretation: - risk-off external shocks usually hurt high-beta A-share growth first - if the shock stabilizes overnight, the strongest domestic growth groups often repair first - on war shocks, test both direct and second-order beneficiaries ### 2. Domestic Policy And Liquidity Layer Check: - `LPR` - PBOC guidance - property and fiscal support headlines - domestic macro expectations Interpretation: - `5Y LPR` matters more for property chain and home appliances - `1Y + 5Y` joint easing broadens the repair base - national or multi-ministry policy language raises mainline probability sharply ### 3. Internal Market Structure Layer Check: - Shanghai Composite, Shenzhen Component, ChiNext, STAR 50, Beijing 50 - up/down breadth - strongest and weakest sector clusters - behavior of core leaders versus high-beta laggards - when possible, brokers versus diversified finance versus banks as a fast sentiment cross-check Interpretation: - better-quality repair starts with resilient growth leaders reclaiming prior close - false repair often shows only defensive strength or isolated single-name squeezes - broker strength supports a more offensive read - bank-only strength usually supports a defensive read ## Distribution And Downgrade Signals Downgrade a board or leader if several of these appear together: - large leader stalls for multiple sessions and volume shrinks - smaller same-board names rotate limit-up one after another - small names surge while the board leader stops making new highs - breadth narrows even while headline excitement stays high Interpretation: - this is often distribution, not healthy continuation - do not confuse "the board still has movers" with "the board is still in accumulation" ## Session Timing Gates ### Pre-open: `08:30-08:55` - Build the overnight shock map. - Decide whether the starting assumption is `mainline`, `independent`, `repair window`, `neutral`, or `defensive`. ### Policy Gate: `09:00` - Check `LPR` on relevant dates. - Re-rank policy-sensitive sectors immediately if the release surprises. ### Auction: `09:20-09:25` - Check which group leads the auction: - growth repair - policy beta - defensive concentration - isolated independent squeezes ### Opening Window: `09:30-10:00` - Confirm whether indices hold key levels. - Confirm whether leadership reclaims prior close. - Decide whether the tape is broad enough for mainline language or only narrow enough for independent-leader language. ### Breadth Window: `10:00-10:30` - Reject moves with poor breadth. - If only a handful of names remain strong, downgrade from mainline to independent or defensive. ### Afternoon Confirmation: `14:00-14:30` - Distinguish sustained repair from technical rebound. - Watch for distribution signs in leaders and low-float satellites. ## Output Template Use: - decision summary - market state: `mainline / independent leader / range-defensive` - strategy mapping for that state - `Base / Bull / Bear` - repair or attack candidates - defensive-only groups - key leaders and invalidation points - opening checklist - `Do / Avoid` FILE:references/opening-window-template.md # Opening Window Template Use this template during the first `30` minutes after the A-share cash open. ## Objective Decide whether the market is showing: - mainline expansion - independent-leader continuation - true repair - policy-led repair - defensive concentration - continuation selloff ## Time Gates ### `09:00` Check: - `LPR` on release days - whether policy-sensitive sectors need to be re-ranked immediately Interpretation: - `5Y LPR` cut: upgrade property chain, home appliances, building materials, brokers - no change: do not assume policy beta will lead on its own ### `09:20-09:25` Check: - auction leadership - whether technology repair or defensive names dominate - whether brokers, diversified finance, or banks are giving a fast sentiment read - whether only one to three names are squeezing without board support Interpretation: - if growth leaders bid well, keep repair scenario live - if several leaders in one chain bid well, keep `mainline` scenario live - if only one to three names detach, classify as `independent leader` first - if oil, coal, banks, and telecom dominate, classify as defensive first ### `09:30-10:00` Check: - index support around key levels - whether resilient leaders can reclaim prior close - whether risk appetite proxies such as brokers improve too Interpretation: - strong repair usually starts with core technology names reclaiming prior close - broad board follow-through is needed before using full `mainline` language - a weak open where only defensives stay green is not healthy repair ### `10:00-10:30` Check: - breadth expansion - whether repair moves broaden beyond the first two or three leaders - whether large leaders are still making progress or already stalling while smaller names rotate Interpretation: - if breadth stays weak, downgrade the move to a technical bounce - if small caps rotate while large leaders stall, consider the board to be entering distribution ## Fast Decision Rules 1. If `tech_repair` beats `defensive_gauge`, favor true repair. 2. If `defensive_gauge` beats `tech_repair`, favor defensive concentration. 3. If `policy_beta` wakes up only after a favorable `LPR`, treat that as policy-led repair. 4. If one to three names squeeze while the board stays mixed, treat the tape as `independent leader`. 5. If several liquid names in one chain confirm with improving breadth, treat the tape as `mainline`. 6. If all groups are weak and indices lose support, treat the tape as continuation selloff. ## Recommended Output - one-line market state - strategy mapping for that state - which group is leading - whether the move is broad or narrow - two or three confirmation names - one invalidation line FILE:references/persona-prompt.md # Persona Prompt You are an A-share discretionary trading decision-maker, not a passive market commentator. Your job is to convert incomplete market information into a concrete next-session game plan. ## Operating Principles 1. Be data-first. Start from verified market structure, sector strength, policy timing, and external shocks. 2. Start with breakout potential. Ask whether the event can break out into public attention and attract large capital. 3. Classify the tape first: `mainline`, `independent leader`, or `range-defensive`. 4. Think in probabilities, not certainties. Always provide a base case, upside case, downside case, and invalidation conditions. 5. Separate explanation from decision. The goal is to decide what matters tomorrow, not to restate everything that happened today. 6. Prefer relative strength over blind mean reversion. In weak tape, the sectors that resisted best are usually better repair candidates than the sectors that fell the most. 7. Distinguish broad repair from defensive concentration. If only oil, coal, banks, telecom, or utilities are strong, that is usually risk aversion, not a healthy market recovery. 8. Prefer understandable logic. If retail can understand the cause quickly and institutions have a reason to stay, follow-through odds improve. 9. For geopolitical or policy shocks, check the second-order beneficiary, not only the crowded first-order trade. 10. Respect policy timing and date anchors. On `LPR` days, treat the `09:00` release as a real branch in the decision tree. On fixed-date events, be aware of pre-positioning and pre-event distribution. 11. Treat pure sentiment gimmicks as temperature checks, not core recommendations. 12. Avoid grand narratives without triggers. Every view must map to a condition the market can confirm or reject. 13. Use exact dates and times whenever timing matters. ## Required Output Shape - One-paragraph decision summary - Market state: `mainline / independent leader / range-defensive` - Strategy mapping for that state - `Base / Bull / Bear` path with conditions and rough probabilities - Sectors most likely to repair first - Sectors likely to stay defensive-only - Key leaders or representative names - A short opening checklist for `09:00`, `09:25`, `09:30-10:00`, and `14:00` - A `do / avoid` section FILE:references/trading-mode-prompt.md # Trading Mode Prompt Use this mode when the task is to prepare for the next A-share session. ## Objective Turn overnight information and current market structure into a next-session execution plan. ## Workflow 1. Pre-open scan: `08:30-08:55` - Check overnight geopolitics, oil, U.S. index direction, rate decisions, and any domestic policy headlines. - Ask first whether the event can break out into broad attention or whether it stays niche. 2. Policy gate: `09:00` - On monthly `LPR` days, treat the release as a major branch point. 3. Auction read: `09:20-09:25` - Identify whether leadership sits in growth repair, defensive concentration, or only one to three isolated names. - Use brokers versus diversified finance versus banks as an auxiliary sentiment gauge when those groups are informative. 4. Opening confirmation: `09:30-10:00` - Validate index support, breadth, and whether high-quality leaders can trade back above prior close. - Classify the tape as `mainline`, `independent leader`, or `range-defensive`. 5. Breadth check: `10:00-10:30` - Reject false rebounds where only a handful of names or defensives are green. - If breadth fails, downgrade from mainline to independent or defensive. 6. Afternoon confirmation: `14:00-14:30` - Decide whether the move is sustaining or reverting into a technical bounce. - Watch for leader stalling plus small-cap rotation, which often signals distribution. ## Decision Rules - If oil stabilizes and growth leaders reclaim prior close, favor technology repair. - If `5Y LPR` is cut, upgrade property chain, home appliances, building materials, and brokers. - If the catalyst is national, global, or multi-ministry and several liquid leaders confirm, upgrade the tape toward `mainline`. - If only one to three names detach without board follow-through, classify as `independent leader`, not full repair. - If only oil, coal, banks, telecom, or utilities lead, classify the tape as defensive. - For geopolitical shocks, check whether the better trade sits in a second-order beneficiary rather than the obvious first-order board. - Do not prioritize the biggest losers for rebound trades unless leadership breadth confirms. - Reassess if the Shanghai Composite loses `4000` or the prior panic low after the open. FILE:scripts/benchmark_sources.py #!/usr/bin/env python3 from __future__ import annotations import argparse import json import time from pathlib import Path from market_data import fetch_index_snapshot, fetch_sector_movers, fetch_tencent_quotes from mx_api import data_query, get_mx_api_key, news_search, stock_screen from news_iterator import DEFAULT_CONFIG, load_config, parse_feed from runtime_config import get_output_dir, require_em_api_key def timed_call(label: str, category: str, func) -> dict: start = time.perf_counter() try: payload = func() elapsed = round(time.perf_counter() - start, 3) details = summarize_payload(payload) return { "label": label, "category": category, "status": "ok", "latency_s": elapsed, "details": details, } except Exception as exc: elapsed = round(time.perf_counter() - start, 3) return { "label": label, "category": category, "status": "error", "latency_s": elapsed, "details": str(exc), } def summarize_payload(payload) -> str: if isinstance(payload, list): return f"items={len(payload)}" if isinstance(payload, dict): if "items" in payload: return f"items={len(payload.get('items') or [])}" if "rows" in payload: return f"rows={len(payload.get('rows') or [])}, total={payload.get('total')}" if "tables" in payload: return f"tables={len(payload.get('tables') or [])}, entities={len(payload.get('entities') or [])}" return f"keys={len(payload)}" return type(payload).__name__ def render_markdown(rows: list[dict]) -> str: lines = ["# Source Benchmark", "", "| Category | Source | Status | Latency(s) | Details |", "| --- | --- | --- | ---: | --- |"] for row in rows: lines.append( f"| {row['category']} | {row['label']} | {row['status']} | {row['latency_s']:.3f} | {row['details']} |" ) return "\n".join(lines) + "\n" def save_outputs(rows: list[dict], output_dir: Path) -> None: output_dir.mkdir(parents=True, exist_ok=True) markdown = render_markdown(rows) (output_dir / "benchmark.md").write_text(markdown, encoding="utf-8") (output_dir / "benchmark.json").write_text(json.dumps(rows, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Benchmark public and MX-enhanced data sources used by the skill.") parser.add_argument("--format", choices=["markdown", "json"], default="markdown") parser.add_argument("--output-dir", help="Optional directory to save benchmark markdown and JSON.") parser.add_argument("--skip-mx", action="store_true", help="Skip MX API calls even if EM_API_KEY is configured.") return parser def main() -> int: require_em_api_key(script_hint="python3 skill/uwillberich/scripts/runtime_config.py set-em-key --stdin") parser = build_parser() args = parser.parse_args() rows = [ timed_call("Eastmoney indices", "public", fetch_index_snapshot), timed_call("Eastmoney sectors", "public", lambda: fetch_sector_movers(limit=5, rising=True)), timed_call("Tencent quotes", "public", lambda: fetch_tencent_quotes(["sz300502", "sh688981", "sh600938"])), ] feed_config = load_config(str(DEFAULT_CONFIG)) first_feed = feed_config.get("feeds", [])[0] if first_feed: rows.append(timed_call(first_feed["label"], "public", lambda: parse_feed(first_feed))) mx_ready = False if not args.skip_mx: try: mx_ready = bool(get_mx_api_key()) except Exception: mx_ready = False if mx_ready: rows.extend( [ timed_call("MX news-search", "mx", lambda: news_search("立讯精密最新资讯", size=6)), timed_call("MX stock-screen", "mx", lambda: stock_screen("A股光模块概念股", page_size=8)), timed_call("MX data-query", "mx", lambda: data_query("浪潮信息最新价总市值收盘价")), ] ) else: rows.extend( [ {"label": "MX news-search", "category": "mx", "status": "skipped", "latency_s": 0.0, "details": "EM_API_KEY not configured or skip requested"}, {"label": "MX stock-screen", "category": "mx", "status": "skipped", "latency_s": 0.0, "details": "EM_API_KEY not configured or skip requested"}, {"label": "MX data-query", "category": "mx", "status": "skipped", "latency_s": 0.0, "details": "EM_API_KEY not configured or skip requested"}, ] ) output_dir = Path(args.output_dir).expanduser() if args.output_dir else get_output_dir("benchmarks") save_outputs(rows, output_dir) if args.format == "json": print(json.dumps(rows, ensure_ascii=False, indent=2)) else: print(render_markdown(rows)) print(f"Saved: {output_dir}") return 0 if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/capital_flow.py #!/usr/bin/env python3 from __future__ import annotations import argparse import json import re from pathlib import Path from mx_api import data_query, stock_screen from runtime_config import require_em_api_key ROOT = Path(__file__).resolve().parents[1] DEFAULT_WATCHLIST = ROOT / "assets" / "default_watchlists.json" MARKET_FLOW_QUERY = "全部A股主力净流入资金今日大单净流入中单净流入小单净流入" TOP_FLOW_QUERIES = { "inflow": "A股主力资金净流入前{limit}股票", "outflow": "A股主力资金净流出前{limit}股票", } def parse_amount_to_yi(value: object) -> float | None: text = str(value or "").strip().replace(",", "") if not text: return None match = re.search(r"-?\d+(?:\.\d+)?", text) if not match: return None number = float(match.group(0)) if "万亿" in text: return round(number * 10000, 2) if "亿" in text: return round(number, 2) if "万" in text: return round(number / 10000, 4) if "元" in text or text.endswith("00"): return round(number / 100000000, 2) return round(number, 2) def subtract_amounts(left: object, right: object) -> float | None: left_value = parse_amount_to_yi(left) right_value = parse_amount_to_yi(right) if left_value is None or right_value is None: return None return round(left_value - right_value, 2) def market_to_symbol(code: str, market: str) -> str: market_code = (market or "").strip().lower() if market_code.startswith("sh"): return f"sh{code}" if market_code.startswith("bj"): return f"bj{code}" return f"sz{code}" def load_watchlist(path: str) -> dict: return json.loads(Path(path).read_text(encoding="utf-8")) def first_column_key(columns: list[dict], title_terms: list[str], fallback_keys: list[str]) -> str: for column in columns: title = str(column.get("title", "")).strip() if any(term in title for term in title_terms): return column.get("key", "") for fallback in fallback_keys: for column in columns: if column.get("key") == fallback: return fallback return "" def build_metric_map(table: dict) -> dict[str, object]: metric_map: dict[str, object] = {} name_map = table.get("nameMap") or {} data_table = table.get("table") or {} for key, label in name_map.items(): if key == "headNameSub": continue values = data_table.get(key) or [] if values: metric_map[str(label)] = values[0] return metric_map def find_metric_value(metric_map: dict[str, object], token_groups: list[list[str]]) -> object | None: for tokens in token_groups: for label, value in metric_map.items(): if all(token in label for token in tokens): return value return None def fetch_market_flow_snapshot() -> dict: result = data_query(MARKET_FLOW_QUERY) target_table = next((table for table in result["tables"] if table.get("entityName") == "全部A股"), None) if not target_table and result["tables"]: target_table = result["tables"][0] if not target_table: return { "label": "未知", "main_net_yi": None, "big_order_net_yi": None, "medium_order_net_yi": None, "small_order_net_yi": None, "as_of": "", } metrics = build_metric_map(target_table) main_net_yi = parse_amount_to_yi( find_metric_value( metrics, [ ["主力净流入资金"], ["主力净额"], ], ) ) big_order_inflow_yi = parse_amount_to_yi(find_metric_value(metrics, [["大单流入资金"], ["大单流入"]])) medium_order_inflow_yi = parse_amount_to_yi(find_metric_value(metrics, [["中单流入资金"], ["中单流入"]])) small_order_inflow_yi = parse_amount_to_yi(find_metric_value(metrics, [["小单流入资金"], ["小单流入"]])) if main_net_yi is None: label = "未知" elif main_net_yi >= 50: label = "强流入" elif main_net_yi > 0: label = "偏流入" elif main_net_yi <= -50: label = "强流出" else: label = "偏流出" return { "label": label, "main_net_yi": main_net_yi, "big_order_inflow_yi": big_order_inflow_yi, "medium_order_inflow_yi": medium_order_inflow_yi, "small_order_inflow_yi": small_order_inflow_yi, "as_of": ((target_table.get("table") or {}).get("headName") or [""])[0], } def fetch_top_main_flows(direction: str, limit: int = 10) -> list[dict]: query = TOP_FLOW_QUERIES[direction].format(limit=limit) result = stock_screen(query, page_no=1, page_size=limit) columns = result["columns"] rows = result["rows"] flow_key = first_column_key(columns, ["主力净额"], ["010000_FLOWZLAMOUNT<70>{2026-03-20}"]) amount_key = first_column_key(columns, ["成交额"], ["010000_TRADING_VOLUMES<70>{2026-03-20}"]) board_key = first_column_key(columns, ["东财行业总分类"], []) concept_key = first_column_key(columns, ["概念"], ["STYLE_CONCEPT"]) items: list[dict] = [] for row in rows[:limit]: code = str(row.get("SECURITY_CODE", "")).strip() market = str(row.get("MARKET_SHORT_NAME", "")).strip() if not code: continue main_flow_yi = parse_amount_to_yi(row.get(flow_key)) trading_amount_yi = parse_amount_to_yi(row.get(amount_key)) item = { "symbol": market_to_symbol(code, market), "code": code, "name": row.get("SECURITY_SHORT_NAME", ""), "market": market, "price": row.get("NEWEST_PRICE"), "change_pct": row.get("CHG"), "main_flow_yi": main_flow_yi, "trading_amount_yi": trading_amount_yi, "board": row.get(board_key, ""), "concept": row.get(concept_key, ""), "direction": direction, "flow_tag": "主力流入榜" if direction == "inflow" else "主力流出榜", } items.append(item) return items def build_flow_lookup(inflow_items: list[dict], outflow_items: list[dict]) -> dict[str, dict]: lookup: dict[str, dict] = {} for item in inflow_items + outflow_items: current = lookup.get(item["symbol"]) if current is None or abs(item.get("main_flow_yi") or 0) > abs(current.get("main_flow_yi") or 0): lookup[item["symbol"]] = dict(item) return lookup def build_group_flow_scoreboard(watchlists: dict, groups: list[str], flow_lookup: dict[str, dict]) -> list[dict]: scoreboard: list[dict] = [] for group in groups: items = watchlists.get(group, []) if not items: continue inflow_hits: list[dict] = [] outflow_hits: list[dict] = [] net_flow_yi = 0.0 for item in items: flow = flow_lookup.get(item["symbol"]) if not flow: continue if flow["direction"] == "inflow": inflow_hits.append(flow) else: outflow_hits.append(flow) net_flow_yi += flow.get("main_flow_yi") or 0.0 if inflow_hits and len(inflow_hits) >= len(outflow_hits): bias = "资金共振" elif outflow_hits and len(outflow_hits) > len(inflow_hits): bias = "资金承压" else: bias = "中性" leaders = inflow_hits if inflow_hits else outflow_hits top_names = "、".join(flow["name"] for flow in leaders[:3]) or "n/a" scoreboard.append( { "group": group, "inflow_hits": len(inflow_hits), "outflow_hits": len(outflow_hits), "net_flow_yi": round(net_flow_yi, 2), "bias": bias, "leaders": top_names, } ) return scoreboard def attach_flow_tags(rows: list[dict], flow_lookup: dict[str, dict]) -> list[dict]: tagged: list[dict] = [] for row in rows: symbol = "" code = str(row.get("code", "")).strip() if code: if code.startswith(("sh", "sz", "bj")): symbol = code elif code[0] in {"6", "9"}: symbol = f"sh{code}" elif code[0] in {"4", "8"}: symbol = f"bj{code}" else: symbol = f"sz{code}" flow = flow_lookup.get(symbol) enriched = dict(row) enriched["flow_tag"] = flow.get("flow_tag", "") if flow else "" enriched["flow_yi"] = flow.get("main_flow_yi") if flow else None tagged.append(enriched) return tagged def render_flow_snapshot(snapshot: dict) -> list[dict]: return [ { "label": snapshot.get("label", ""), "main_net_yi": snapshot.get("main_net_yi"), "big_order_inflow_yi": snapshot.get("big_order_inflow_yi"), "medium_order_inflow_yi": snapshot.get("medium_order_inflow_yi"), "small_order_inflow_yi": snapshot.get("small_order_inflow_yi"), "as_of": snapshot.get("as_of", ""), } ] def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Capital-flow monitor for uwillberich.") parser.add_argument( "--watchlist", default=str(DEFAULT_WATCHLIST), help="Watchlist JSON path.", ) parser.add_argument( "--groups", nargs="+", default=["tech_repair", "defensive_gauge"], help="Watchlist groups to intersect with top-flow leaderboards.", ) parser.add_argument("--limit", type=int, default=10, help="Top inflow/outflow rows to fetch.") parser.add_argument("--format", choices=["markdown", "json"], default="markdown") return parser def main() -> int: require_em_api_key(script_hint="python3 skill/uwillberich/scripts/runtime_config.py set-em-key --stdin") args = build_parser().parse_args() watchlists = load_watchlist(args.watchlist) snapshot = fetch_market_flow_snapshot() inflow_items = fetch_top_main_flows("inflow", limit=args.limit) outflow_items = fetch_top_main_flows("outflow", limit=args.limit) flow_lookup = build_flow_lookup(inflow_items, outflow_items) group_flow = build_group_flow_scoreboard(watchlists, args.groups, flow_lookup) payload = { "market_flow": snapshot, "top_inflow": inflow_items, "top_outflow": outflow_items, "group_flow": group_flow, } if args.format == "json": print(json.dumps(payload, ensure_ascii=False, indent=2)) return 0 print("# Capital Flow") print() print(json.dumps(payload, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/fetch_market_snapshot.py #!/usr/bin/env python3 from __future__ import annotations import argparse import json from market_data import fetch_index_snapshot, fetch_sector_movers, format_markdown_table def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Fetch A-share index and sector breadth snapshots.") parser.add_argument("--limit", type=int, default=10, help="Number of top and bottom sectors to return.") parser.add_argument("--format", choices=["json", "markdown"], default="markdown") return parser def main() -> None: parser = build_parser() args = parser.parse_args() indices = fetch_index_snapshot() leaders = fetch_sector_movers(limit=args.limit, rising=True) laggards = fetch_sector_movers(limit=args.limit, rising=False) if args.format == "json": payload = {"indices": indices, "leaders": leaders, "laggards": laggards} print(json.dumps(payload, ensure_ascii=False, indent=2)) return print("## Indices") print( format_markdown_table( indices, [ ("Name", "name"), ("Price", "price"), ("Chg%", "change_pct"), ("Up", "up_count"), ("Down", "down_count"), ], ) ) print("\n## Top Sectors") print( format_markdown_table( leaders, [ ("Sector", "name"), ("Chg%", "change_pct"), ("Leader", "leader"), ("Up", "up_count"), ("Down", "down_count"), ], ) ) print("\n## Bottom Sectors") print( format_markdown_table( laggards, [ ("Sector", "name"), ("Chg%", "change_pct"), ("Leader", "leader"), ("Up", "up_count"), ("Down", "down_count"), ], ) ) if __name__ == "__main__": main() FILE:scripts/fetch_quotes.py #!/usr/bin/env python3 from __future__ import annotations import argparse import json from market_data import fetch_tencent_quotes, format_markdown_table def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Fetch Tencent quote snapshots for A-share watchlists.") parser.add_argument("symbols", nargs="+", help="Symbols such as sz300502 sh688981 sh600938") parser.add_argument("--format", choices=["json", "markdown"], default="markdown") return parser def main() -> None: parser = build_parser() args = parser.parse_args() quotes = fetch_tencent_quotes(args.symbols) if args.format == "json": print(json.dumps(quotes, ensure_ascii=False, indent=2)) return columns = [ ("Name", "name"), ("Code", "code"), ("Price", "price"), ("Chg%", "change_pct"), ("High", "high"), ("Low", "low"), ("Amount(100m)", "amount_100m"), ] print(format_markdown_table(quotes, columns)) if __name__ == "__main__": main() FILE:scripts/industry_chain.py #!/usr/bin/env python3 from __future__ import annotations import argparse import json from pathlib import Path from mx_api import stock_screen from runtime_config import require_em_api_key ROOT = Path(__file__).resolve().parents[1] DEFAULT_WATCHLIST = ROOT / "assets" / "default_watchlists.json" DEFAULT_CHAIN_CONFIG = ROOT / "assets" / "industry_chains.json" DEFAULT_EVENT_WATCHLIST = Path.home() / ".uwillberich" / "news-iterator" / "event_watchlists.json" def load_json(path: str) -> dict: return json.loads(Path(path).read_text(encoding="utf-8")) def merge_item_details(existing: dict, incoming: dict) -> dict: merged = dict(existing) for key, value in incoming.items(): if not merged.get(key) and value: merged[key] = value return merged def build_symbol_index(base_watchlists: dict) -> dict[str, dict]: symbol_index: dict[str, dict] = {} for items in base_watchlists.values(): for item in items: symbol = item["symbol"] if symbol in symbol_index: symbol_index[symbol] = merge_item_details(symbol_index[symbol], item) else: symbol_index[symbol] = dict(item) return symbol_index def market_to_symbol(code: str, market: str) -> str: market_code = (market or "").strip().lower() if market_code.startswith("sh"): return f"sh{code}" if market_code.startswith("bj"): return f"bj{code}" return f"sz{code}" def select_chain_themes(event_payload: dict, selected_groups: list[str], chain_config: dict, max_themes: int = 3) -> list[dict]: score_map: dict[str, int] = {} reason_map: dict[str, list[str]] = {} theme_map = {theme["id"]: theme for theme in chain_config.get("themes", [])} group_theme_hints = chain_config.get("group_theme_hints", {}) def bump(theme_id: str, points: int, reason: str) -> None: score_map[theme_id] = score_map.get(theme_id, 0) + points reason_map.setdefault(theme_id, []) if reason not in reason_map[theme_id]: reason_map[theme_id].append(reason) for group in selected_groups: for theme_id in group_theme_hints.get(group, []): bump(theme_id, 3, f"group:{group}") summary = event_payload.get("summary", []) default_report_groups = event_payload.get("default_report_groups", []) for theme in chain_config.get("themes", []): theme_id = theme["id"] preferred_groups = set(theme.get("preferred_groups", [])) for group in default_report_groups: if group in preferred_groups: bump(theme_id, 2, f"event_group:{group}") trigger_terms = [term.lower() for term in theme.get("triggers", [])] for item in summary: if item.get("category") in theme.get("categories", []): category_points = max(3, int((int(item.get("total_score", 0)) + 2) / 3)) bump(theme_id, category_points, f"category:{item.get('category')}") for keyword in item.get("top_keywords", []): lower_keyword = str(keyword).lower() if any(term in lower_keyword or lower_keyword in term for term in trigger_terms): bump(theme_id, max(2, item.get("alert_count", 1)), f"keyword:{keyword}") ranked = sorted( ( { "id": theme_id, "label": theme_map[theme_id]["label"], "query": theme_map[theme_id]["query"], "score": score, "reasons": reason_map.get(theme_id, []), } for theme_id, score in score_map.items() if score > 0 and theme_id in theme_map ), key=lambda item: (-item["score"], item["id"]), ) return ranked[:max_themes] def first_column_key(columns: list[dict], title_terms: list[str], fallback_keys: list[str]) -> str: for column in columns: title = str(column.get("title", "")).strip() if any(term in title for term in title_terms): return column.get("key", "") for fallback in fallback_keys: for column in columns: if column.get("key") == fallback: return fallback return "" def build_chain_item(theme: dict, row: dict, symbol_index: dict[str, dict], columns: list[dict], theme_score: int) -> dict: code = str(row.get("SECURITY_CODE", "")).strip() market = str(row.get("MARKET_SHORT_NAME", "")).strip() symbol = market_to_symbol(code, market) base = symbol_index.get(symbol, {}) board_key = first_column_key(columns, ["东财行业总分类"], []) concept_key = first_column_key(columns, ["概念"], ["STYLE_CONCEPT"]) flow_key = first_column_key(columns, ["主力净额"], []) role = base.get("role") or base.get("chain_role") or f"{theme['label']}观察" board = str(row.get(board_key, "")).strip() concept = str(row.get(concept_key, "")).strip() driver_parts = ["产业链", theme["label"]] if board: driver_parts.append(board) elif concept: driver_parts.append(concept[:32]) return { "symbol": symbol, "name": row.get("SECURITY_SHORT_NAME", ""), "role": role, "event_score": theme_score, "trigger_count": 1, "event_driver": " | ".join(driver_parts), "strong_signal": base.get("strong_signal") or theme.get("strong_signal", ""), "weak_signal": base.get("weak_signal") or theme.get("weak_signal", ""), "chain_theme": theme["label"], "chain_query": theme["query"], "flow_hint": row.get(flow_key, ""), } def fetch_chain_group(theme: dict, symbol_index: dict[str, dict], limit: int, theme_score: int) -> list[dict]: result = stock_screen(theme["query"], page_no=1, page_size=max(limit, 10)) items: list[dict] = [] seen: set[str] = set() for row in result["rows"]: code = str(row.get("SECURITY_CODE", "")).strip() market = str(row.get("MARKET_SHORT_NAME", "")).strip() if not code: continue symbol = market_to_symbol(code, market) if symbol in seen: continue seen.add(symbol) items.append(build_chain_item(theme, row, symbol_index, result["columns"], theme_score)) if len(items) >= limit: break return items def enrich_event_payload_with_chain_focus( event_payload: dict, base_watchlists: dict, selected_groups: list[str] | None = None, chain_config_path: str | None = None, max_themes: int = 3, limit: int = 6, ) -> dict: if not event_payload: return {} chain_config = load_json(chain_config_path or str(DEFAULT_CHAIN_CONFIG)) symbol_index = build_symbol_index(base_watchlists) selected_themes = select_chain_themes(event_payload, selected_groups or [], chain_config, max_themes=max_themes) if not selected_themes: return event_payload theme_map = {theme["id"]: theme for theme in chain_config.get("themes", [])} enriched = { **event_payload, "groups": dict(event_payload.get("groups", {})), "default_report_groups": list(event_payload.get("default_report_groups", [])), "chain_summary": [], } existing_groups = set(enriched["default_report_groups"]) for selected in selected_themes: theme = theme_map[selected["id"]] try: items = fetch_chain_group(theme, symbol_index, limit=limit, theme_score=selected["score"]) except Exception as exc: chain_errors = list(enriched.get("chain_errors", [])) chain_errors.append({"theme": theme["label"], "error": str(exc)}) enriched["chain_errors"] = chain_errors continue if not items: continue group_name = f"chain_focus_{theme['id']}" enriched["groups"][group_name] = items if group_name not in existing_groups: enriched["default_report_groups"].append(group_name) existing_groups.add(group_name) enriched["chain_summary"].append( { "group": group_name, "theme": theme["label"], "score": selected["score"], "query": theme["query"], "reasons": selected.get("reasons", []), } ) return enriched def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Build industry-chain focus pools from events and watchlists.") parser.add_argument("--watchlist", default=str(DEFAULT_WATCHLIST), help="Base watchlist JSON path.") parser.add_argument( "--event-watchlist", default=str(DEFAULT_EVENT_WATCHLIST), help="Path to event_watchlists.json generated by news_iterator.", ) parser.add_argument("--chain-config", default=str(DEFAULT_CHAIN_CONFIG), help="Industry-chain config JSON path.") parser.add_argument("--groups", nargs="+", default=["tech_repair", "defensive_gauge"], help="Current desk groups.") parser.add_argument("--limit", type=int, default=6, help="Names per chain theme.") parser.add_argument("--max-themes", type=int, default=3, help="Maximum number of chain themes.") parser.add_argument("--format", choices=["markdown", "json"], default="markdown") return parser def main() -> int: require_em_api_key(script_hint="python3 skill/uwillberich/scripts/runtime_config.py set-em-key --stdin") args = build_parser().parse_args() base_watchlists = load_json(args.watchlist) event_payload = load_json(args.event_watchlist) enriched = enrich_event_payload_with_chain_focus( event_payload, base_watchlists, selected_groups=args.groups, chain_config_path=args.chain_config, max_themes=args.max_themes, limit=args.limit, ) if args.format == "json": print(json.dumps(enriched, ensure_ascii=False, indent=2)) return 0 print("# Industry Chain Focus") print() print(json.dumps(enriched.get("chain_summary", []), ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/install_news_iterator_launchd.py #!/usr/bin/env python3 from __future__ import annotations import argparse import os import plistlib import subprocess import sys from pathlib import Path from runtime_config import load_runtime_env, require_em_api_key ROOT = Path(__file__).resolve().parents[1] DEFAULT_STATE_DIR = Path.home() / ".uwillberich" / "news-iterator" DEFAULT_LABEL = "com.tingchi.uwillberich-news-iterator" DEFAULT_PLIST = Path.home() / "Library" / "LaunchAgents" / f"{DEFAULT_LABEL}.plist" load_runtime_env() require_em_api_key(script_hint="python3 skill/uwillberich/scripts/runtime_config.py set-em-key --stdin") def run_command(args: list[str], check: bool) -> subprocess.CompletedProcess[str]: return subprocess.run(args, text=True, capture_output=True, check=check) def build_plist(interval_seconds: int, state_dir: Path, python_bin: str) -> dict: state_dir.mkdir(parents=True, exist_ok=True) load_runtime_env() plist = { "Label": DEFAULT_LABEL, "ProgramArguments": [ python_bin, str(ROOT / "scripts" / "news_iterator.py"), "--state-dir", str(state_dir), "poll", ], "RunAtLoad": True, "StartInterval": interval_seconds, "WorkingDirectory": str(ROOT), "StandardOutPath": str(state_dir / "launchd.out.log"), "StandardErrorPath": str(state_dir / "launchd.err.log"), } env_vars = {} runtime_env_path = os.environ.get("UWILLBERICH_RUNTIME_ENV") or os.environ.get("A_SHARE_RUNTIME_ENV") if runtime_env_path: env_vars["UWILLBERICH_RUNTIME_ENV"] = runtime_env_path if env_vars: plist["EnvironmentVariables"] = env_vars return plist def unload_if_present(plist_path: Path) -> None: domain = f"gui/{os.getuid()}" run_command(["launchctl", "bootout", domain, str(plist_path)], check=False) def install(args: argparse.Namespace) -> int: plist_path = Path(args.plist_path) plist_path.parent.mkdir(parents=True, exist_ok=True) state_dir = Path(args.state_dir) plist = build_plist(args.interval_seconds, state_dir, args.python_bin) with plist_path.open("wb") as handle: plistlib.dump(plist, handle) unload_if_present(plist_path) domain = f"gui/{os.getuid()}" run_command(["launchctl", "bootstrap", domain, str(plist_path)], check=True) run_command(["launchctl", "kickstart", "-k", f"{domain}/{DEFAULT_LABEL}"], check=False) print(f"installed: {plist_path}") print(f"state_dir: {state_dir}") print(f"interval_seconds: {args.interval_seconds}") return 0 def uninstall(args: argparse.Namespace) -> int: plist_path = Path(args.plist_path) if plist_path.exists(): unload_if_present(plist_path) plist_path.unlink() print(f"removed: {plist_path}") else: print(f"not found: {plist_path}") return 0 def status(args: argparse.Namespace) -> int: plist_path = Path(args.plist_path) print(f"plist: {plist_path}") print(f"exists: {plist_path.exists()}") if not plist_path.exists(): return 0 result = run_command(["launchctl", "print", f"gui/{os.getuid()}/{DEFAULT_LABEL}"], check=False) if result.returncode == 0: print(result.stdout.strip()) else: print(result.stderr.strip() or result.stdout.strip()) return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Install the news iterator as a launchd agent on macOS.") subparsers = parser.add_subparsers(dest="command", required=True) install_parser = subparsers.add_parser("install", help="Install and load the launchd job.") install_parser.add_argument("--interval-seconds", type=int, default=300, help="Polling interval.") install_parser.add_argument("--state-dir", default=str(DEFAULT_STATE_DIR), help="State directory.") install_parser.add_argument("--plist-path", default=str(DEFAULT_PLIST), help="LaunchAgent plist path.") install_parser.add_argument("--python-bin", default=sys.executable, help="Python interpreter path.") install_parser.set_defaults(func=install) uninstall_parser = subparsers.add_parser("uninstall", help="Unload and remove the launchd job.") uninstall_parser.add_argument("--plist-path", default=str(DEFAULT_PLIST), help="LaunchAgent plist path.") uninstall_parser.set_defaults(func=uninstall) status_parser = subparsers.add_parser("status", help="Show launchd job status.") status_parser.add_argument("--plist-path", default=str(DEFAULT_PLIST), help="LaunchAgent plist path.") status_parser.set_defaults(func=status) return parser def main() -> int: parser = build_parser() args = parser.parse_args() return args.func(args) if __name__ == "__main__": sys.exit(main()) FILE:scripts/market_data.py #!/usr/bin/env python3 from __future__ import annotations import json import urllib.parse import urllib.request from typing import Iterable from runtime_config import load_runtime_env, require_em_api_key DEFAULT_HEADERS = {"User-Agent": "Mozilla/5.0"} load_runtime_env() require_em_api_key(script_hint="python3 skill/uwillberich/scripts/runtime_config.py set-em-key --stdin") def _get_text(url: str, encoding: str = "utf-8") -> str: request = urllib.request.Request(url, headers=DEFAULT_HEADERS) with urllib.request.urlopen(request, timeout=10) as response: return response.read().decode(encoding, errors="replace") def _get_json(url: str) -> dict: return json.loads(_get_text(url)) def _to_float(value: str) -> float | None: try: return float(value) except (TypeError, ValueError): return None def _to_int(value: str) -> int | None: try: return int(float(value)) except (TypeError, ValueError): return None def fetch_tencent_quotes(symbols: Iterable[str]) -> list[dict]: symbol_list = [symbol.strip() for symbol in symbols if symbol.strip()] if not symbol_list: return [] url = "https://qt.gtimg.cn/q=" + ",".join(symbol_list) raw = _get_text(url, encoding="gbk") quotes: list[dict] = [] for line in raw.strip().split(";"): if not line or '="' not in line: continue _, value = line.split('="', 1) fields = value.rstrip('"').split("~") if len(fields) < 38: continue amount = _to_float(fields[37]) quotes.append( { "name": fields[1], "code": fields[2], "price": _to_float(fields[3]), "prev_close": _to_float(fields[4]), "open": _to_float(fields[5]), "timestamp": fields[30], "change": _to_float(fields[31]), "change_pct": _to_float(fields[32]), "high": _to_float(fields[33]), "low": _to_float(fields[34]), "volume_lots": _to_int(fields[36]), "amount": amount, "amount_100m": round(amount / 10000, 2) if amount is not None else None, } ) return quotes DEFAULT_INDICES = { "1.000001": "上证指数", "0.399001": "深证成指", "0.399006": "创业板指", "1.000300": "沪深300", "1.000688": "科创50", "0.899050": "北证50", } def fetch_index_snapshot(secids: dict[str, str] | None = None) -> list[dict]: secids = secids or DEFAULT_INDICES params = { "fltt": "2", "invt": "2", "fields": "f12,f14,f2,f3,f4,f104,f105", "secids": ",".join(secids.keys()), } url = "https://push2.eastmoney.com/api/qt/ulist.np/get?" + urllib.parse.urlencode(params) payload = _get_json(url) items = payload.get("data", {}).get("diff", []) snapshot: list[dict] = [] for item in items: snapshot.append( { "code": item.get("f12"), "name": item.get("f14"), "price": item.get("f2"), "change_pct": item.get("f3"), "change": item.get("f4"), "up_count": item.get("f104"), "down_count": item.get("f105"), } ) return snapshot def fetch_sector_movers(limit: int = 10, rising: bool = False) -> list[dict]: params = { "pn": "1", "pz": str(limit), "po": "1" if rising else "0", "np": "1", "fltt": "2", "invt": "2", "fid": "f3", "fs": "m:90+t:3", "fields": "f12,f14,f2,f3,f4,f104,f105,f128", } url = "https://push2.eastmoney.com/api/qt/clist/get?" + urllib.parse.urlencode(params) payload = _get_json(url) items = payload.get("data", {}).get("diff", []) sectors: list[dict] = [] for item in items: sectors.append( { "code": item.get("f12"), "name": item.get("f14"), "price": item.get("f2"), "change_pct": item.get("f3"), "change": item.get("f4"), "up_count": item.get("f104"), "down_count": item.get("f105"), "leader": item.get("f128"), } ) return sectors def format_markdown_table(rows: list[dict], columns: list[tuple[str, str]]) -> str: header = "| " + " | ".join(title for title, _ in columns) + " |" separator = "| " + " | ".join(["---"] * len(columns)) + " |" body = [] for row in rows: values = [] for _, key in columns: value = row.get(key, "") if isinstance(value, float): if value.is_integer(): value = int(value) else: value = f"{value:.2f}" values.append(str(value).replace("|", "\\|").replace("\n", " ")) body.append("| " + " | ".join(values) + " |") return "\n".join([header, separator, *body]) FILE:scripts/market_sentiment.py #!/usr/bin/env python3 from __future__ import annotations import argparse import json from capital_flow import fetch_market_flow_snapshot from market_data import fetch_index_snapshot, fetch_sector_movers def safe_avg(values: list[float | int | None]) -> float | None: usable = [float(value) for value in values if value is not None] if not usable: return None return round(sum(usable) / len(usable), 2) def compute_breadth(indices: list[dict]) -> dict: tracked = [item for item in indices if item.get("name") in {"上证指数", "深证成指"}] up = sum(int(item.get("up_count") or 0) for item in tracked) down = sum(int(item.get("down_count") or 0) for item in tracked) total = up + down ratio = round(up / total, 4) if total else None return {"up": up, "down": down, "ratio": ratio} def score_breadth(ratio: float | None) -> int: if ratio is None: return 0 if ratio >= 0.65: return 2 if ratio >= 0.52: return 1 if ratio <= 0.32: return -2 if ratio <= 0.45: return -1 return 0 def score_main_flow(main_net_yi: float | None) -> int: if main_net_yi is None: return 0 if main_net_yi >= 80: return 2 if main_net_yi > 0: return 1 if main_net_yi <= -80: return -2 if main_net_yi < 0: return -1 return 0 def classify_group_tone(group_flow_rows: list[dict]) -> tuple[str, int]: tone = "mixed" score = 0 by_name = {row["group"]: row for row in group_flow_rows} tech = by_name.get("tech_repair", {}) defensive = by_name.get("defensive_gauge", {}) policy = by_name.get("policy_beta", {}) tech_net = float(tech.get("net_flow_yi") or 0) defensive_net = float(defensive.get("net_flow_yi") or 0) policy_net = float(policy.get("net_flow_yi") or 0) if defensive_net > 0 and tech_net <= 0: tone = "defensive" score = -1 elif tech_net > 0 and defensive_net <= 0: tone = "growth" score = 1 elif policy_net > 0 and tech_net >= 0: tone = "policy-growth" score = 1 return tone, score def build_sentiment_snapshot(group_flow_rows: list[dict] | None = None) -> dict: indices = fetch_index_snapshot() top_sectors = fetch_sector_movers(limit=5, rising=True) bottom_sectors = fetch_sector_movers(limit=5, rising=False) flow_snapshot = fetch_market_flow_snapshot() breadth = compute_breadth(indices) top_avg = safe_avg([item.get("change_pct") for item in top_sectors]) bottom_avg = safe_avg([item.get("change_pct") for item in bottom_sectors]) breadth_score = score_breadth(breadth["ratio"]) flow_score = score_main_flow(flow_snapshot.get("main_net_yi")) dispersion_score = 0 if top_avg is not None and bottom_avg is not None: if top_avg >= 2.5 and bottom_avg <= -2.5: dispersion_score = -1 elif top_avg >= 2.0 and bottom_avg >= -1.0: dispersion_score = 1 group_tone, group_score = classify_group_tone(group_flow_rows or []) total_score = breadth_score + flow_score + dispersion_score + group_score if group_tone == "defensive" and breadth_score <= 0: label = "抱团行情" read = "资金更集中在防御和高确定性方向，广度没有同步改善。" elif group_tone == "growth" and total_score >= 1: label = "科技修复" read = "成长方向开始获得资金确认，但仍要看扩散是否持续。" elif total_score >= 2: label = "修复扩散" read = "广度、板块和资金同步改善，情绪修复质量较高。" elif total_score <= -2: label = "分化偏弱" read = "广度和主力资金都偏弱，反弹更像局部脉冲。" else: label = "分化震荡" read = "结构性轮动还在，市场没有给出统一方向。" components = [ {"component": "市场广度", "score": breadth_score, "detail": f"{breadth['up']} / {breadth['down']}"}, { "component": "主力资金", "score": flow_score, "detail": f"{flow_snapshot.get('main_net_yi')}亿" if flow_snapshot.get("main_net_yi") is not None else "n/a", }, { "component": "板块扩散", "score": dispersion_score, "detail": f"强势板块均值 {top_avg}%，弱势板块均值 {bottom_avg}%", }, {"component": "观察池风格", "score": group_score, "detail": group_tone}, ] return { "label": label, "read": read, "score": total_score, "breadth": breadth, "market_flow": flow_snapshot, "top_sector_avg": top_avg, "bottom_sector_avg": bottom_avg, "group_tone": group_tone, "components": components, } def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Market sentiment snapshot for A-share Decision Desk.") parser.add_argument("--format", choices=["markdown", "json"], default="markdown") return parser def main() -> int: args = build_parser().parse_args() snapshot = build_sentiment_snapshot() if args.format == "json": print(json.dumps(snapshot, ensure_ascii=False, indent=2)) return 0 print("# Market Sentiment") print() print(f"- state: {snapshot['label']}") print(f"- read: {snapshot['read']}") print(json.dumps(snapshot["components"], ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/morning_brief.py #!/usr/bin/env python3 from __future__ import annotations import argparse import json from pathlib import Path from capital_flow import ( attach_flow_tags, build_flow_lookup, build_group_flow_scoreboard, fetch_market_flow_snapshot, fetch_top_main_flows, render_flow_snapshot, ) from industry_chain import enrich_event_payload_with_chain_focus from market_data import fetch_index_snapshot, fetch_sector_movers, fetch_tencent_quotes, format_markdown_table from market_sentiment import build_sentiment_snapshot ROOT = Path(__file__).resolve().parents[1] DEFAULT_WATCHLIST = ROOT / "assets" / "default_watchlists.json" DEFAULT_EVENT_WATCHLIST = Path.home() / ".uwillberich" / "news-iterator" / "event_watchlists.json" EVENT_CATEGORY_ORDER = ["huge_conflict", "huge_future", "huge_name_release"] CATEGORY_LABELS = { "huge_conflict": "巨大冲突", "huge_future": "巨大前景", "huge_name_release": "巨头名人", } SIGNAL_LABELS = {"high": "高", "medium": "中", "low": "低"} KEYWORD_LABELS = { "war": "战争", "oil": "原油", "energy": "能源", "chips": "芯片", "chip": "芯片", "robots": "机器人", "robot": "机器人", "launch": "发布", "launches": "发布", "announces": "宣布", "announce": "宣布", "unveils": "亮相", "unveil": "亮相", "data center": "数据中心", } def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Build a simple A-share morning brief from default watchlists.") parser.add_argument( "--watchlist", default=str(DEFAULT_WATCHLIST), help="Path to a watchlist JSON file. Defaults to the bundled watchlist.", ) parser.add_argument( "--groups", nargs="+", default=["core10"], help="Watchlist groups to print, for example: core10 tech_repair defensive_gauge", ) parser.add_argument( "--event-watchlist", default=str(DEFAULT_EVENT_WATCHLIST), help="Path to dynamic event-driven watchlists JSON.", ) parser.add_argument( "--skip-event-pools", action="store_true", help="Do not append event-driven watchlists from the news iterator state.", ) parser.add_argument( "--skip-capital-flow", action="store_true", help="Do not append main-force capital-flow sections.", ) parser.add_argument( "--skip-sentiment", action="store_true", help="Do not append the market-sentiment snapshot.", ) parser.add_argument( "--skip-industry-chain", action="store_true", help="Do not enrich event pools with chain-focus groups.", ) return parser def load_watchlist(path: str) -> dict: return json.loads(Path(path).read_text(encoding="utf-8")) def load_event_payload(path: str) -> dict: event_path = Path(path) if not event_path.exists(): return {} return json.loads(event_path.read_text(encoding="utf-8")) def category_display_name(category: str) -> str: return CATEGORY_LABELS.get(category, category) def signal_display_name(signal: str) -> str: return SIGNAL_LABELS.get(signal, signal) def format_keyword_list(keywords: list[str]) -> str: if not keywords: return "n/a" return ", ".join(KEYWORD_LABELS.get(keyword, keyword) for keyword in keywords) def build_rows(items: list[dict], quotes: list[dict]) -> list[dict]: quote_map = {quote["code"]: quote for quote in quotes} rows: list[dict] = [] for item in items: code = item["symbol"][2:] quote = quote_map.get(code) if not quote: continue rows.append( { "name": quote["name"], "code": quote["code"], "role": item["role"], "price": quote["price"], "change_pct": quote["change_pct"], "high": quote["high"], "low": quote["low"], "amount_100m": quote["amount_100m"], "event_score": item.get("event_score"), "trigger_count": item.get("trigger_count"), "event_driver": item.get("event_driver", ""), } ) return rows def render_watchlist_table(rows: list[dict], is_event: bool) -> str: columns = [ ("Name", "name"), ("Code", "code"), ("Role", "role"), ] if is_event: columns.extend( [ ("EventScore", "event_score"), ("Triggers", "trigger_count"), ("Driver", "event_driver"), ] ) columns.extend( [ ("FlowTag", "flow_tag"), ("Flow(亿)", "flow_yi"), ("Price", "price"), ("Chg%", "change_pct"), ("High", "high"), ("Low", "low"), ("Amount(100m)", "amount_100m"), ] ) return format_markdown_table(rows, columns) def render_event_summary(payload: dict) -> None: summary = payload.get("summary", []) if not summary: return rows = [ { "category": category_display_name(item["category"]), "alert_count": item["alert_count"], "total_score": item["total_score"], "top_keywords": format_keyword_list(item.get("top_keywords", [])), } for item in summary ] print("\n## 事件驱动层总结") print( format_markdown_table( rows, [ ("类别", "category"), ("条数", "alert_count"), ("总分", "total_score"), ("高频关键词", "top_keywords"), ], ) ) def render_event_top_alerts(payload: dict) -> None: top_alerts = payload.get("top_alerts", {}) if not top_alerts: return print("\n## 事件信息源链接") for category in EVENT_CATEGORY_ORDER: items = top_alerts.get(category, []) if not items: continue print(f"\n### {category_display_name(category)} Top 10 信息源") for index, item in enumerate(items, start=1): print(f"{index}. [{item['title']}]({item['link']})") print( f" - 来源: {item['source']} | 信号: `{signal_display_name(item['signal'])}` | 分值: `{item['score']}`" ) print(f" - 实体: {', '.join(item.get('entities', [])) or 'n/a'}") print(f" - 关键词: {format_keyword_list(item.get('keywords', []))}") def render_chain_summary(payload: dict) -> None: summary = payload.get("chain_summary", []) if not summary: return rows = [ { "theme": item["theme"], "score": item["score"], "group": item["group"], "reasons": " / ".join(item.get("reasons", [])[:3]) or "n/a", } for item in summary ] print("\n## Industry Chain Focus") print( format_markdown_table( rows, [ ("Theme", "theme"), ("Score", "score"), ("Group", "group"), ("Reasons", "reasons"), ], ) ) def main() -> None: parser = build_parser() args = parser.parse_args() watchlist = load_watchlist(args.watchlist) event_payload = {} if args.skip_event_pools else load_event_payload(args.event_watchlist) if event_payload and not args.skip_industry_chain: event_payload = enrich_event_payload_with_chain_focus( event_payload, watchlist, selected_groups=args.groups, ) event_groups = event_payload.get("groups", {}) selected_groups = [group for group in args.groups if group in watchlist] selected_event_groups = [group for group in args.groups if group in event_groups] if not selected_event_groups and event_groups: selected_event_groups = event_payload.get("default_report_groups", []) selected_event_groups = list(dict.fromkeys(selected_event_groups)) print("# A-Share Morning Brief") print("\n## Indices") print( format_markdown_table( fetch_index_snapshot(), [ ("Name", "name"), ("Price", "price"), ("Chg%", "change_pct"), ("Up", "up_count"), ("Down", "down_count"), ], ) ) print("\n## Top Sectors") print( format_markdown_table( fetch_sector_movers(limit=5, rising=True), [("Sector", "name"), ("Chg%", "change_pct"), ("Leader", "leader")], ) ) print("\n## Bottom Sectors") print( format_markdown_table( fetch_sector_movers(limit=5, rising=False), [("Sector", "name"), ("Chg%", "change_pct"), ("Leader", "leader")], ) ) flow_lookup: dict[str, dict] = {} group_flow_rows: list[dict] = [] if not args.skip_capital_flow: market_flow = fetch_market_flow_snapshot() inflow_items = fetch_top_main_flows("inflow", limit=8) outflow_items = fetch_top_main_flows("outflow", limit=8) flow_lookup = build_flow_lookup(inflow_items, outflow_items) group_flow_rows = build_group_flow_scoreboard(watchlist, selected_groups, flow_lookup) print("\n## Capital Flow Snapshot") print( format_markdown_table( render_flow_snapshot(market_flow), [ ("State", "label"), ("MainNet(亿)", "main_net_yi"), ("BigInflow(亿)", "big_order_inflow_yi"), ("MediumInflow(亿)", "medium_order_inflow_yi"), ("SmallInflow(亿)", "small_order_inflow_yi"), ("As Of", "as_of"), ], ) ) print("\n## Top Main-Force Inflow") print( format_markdown_table( inflow_items[:5], [ ("Name", "name"), ("Code", "code"), ("Chg%", "change_pct"), ("MainFlow(亿)", "main_flow_yi"), ("Board", "board"), ], ) ) print("\n## Top Main-Force Outflow") print( format_markdown_table( outflow_items[:5], [ ("Name", "name"), ("Code", "code"), ("Chg%", "change_pct"), ("MainFlow(亿)", "main_flow_yi"), ("Board", "board"), ], ) ) if group_flow_rows: print("\n## Watchlist Flow Resonance") print( format_markdown_table( group_flow_rows, [ ("Group", "group"), ("InflowHits", "inflow_hits"), ("OutflowHits", "outflow_hits"), ("NetFlow(亿)", "net_flow_yi"), ("Bias", "bias"), ("Leaders", "leaders"), ], ) ) if not args.skip_sentiment: sentiment = build_sentiment_snapshot(group_flow_rows=group_flow_rows) print("\n## Sentiment Snapshot") print(f"- state: {sentiment['label']}") print(f"- read: {sentiment['read']}") print( format_markdown_table( sentiment["components"], [ ("Component", "component"), ("Score", "score"), ("Detail", "detail"), ], ) ) for group in selected_groups: items = watchlist[group] quotes = fetch_tencent_quotes(item["symbol"] for item in items) rows = attach_flow_tags(build_rows(items, quotes), flow_lookup) print(f"\n## Watchlist: {group}") print(render_watchlist_table(rows, is_event=False)) if event_groups and selected_event_groups: render_event_summary(event_payload) render_event_top_alerts(event_payload) render_chain_summary(event_payload) for group in selected_event_groups: items = event_groups.get(group, []) if not items: continue quotes = fetch_tencent_quotes(item["symbol"] for item in items) rows = attach_flow_tags(build_rows(items, quotes), flow_lookup) print(f"\n## Event Watchlist: {group}") print(render_watchlist_table(rows, is_event=True)) if __name__ == "__main__": main() FILE:scripts/mx_api.py #!/usr/bin/env python3 from __future__ import annotations import csv import json import os import urllib.request from pathlib import Path from runtime_config import load_runtime_env, require_em_api_key MX_BASE_URL = "https://mkapi2.dfcfs.com/finskillshub/api/claw" DEFAULT_HEADERS = {"Content-Type": "application/json"} load_runtime_env() def get_mx_api_key() -> str: return require_em_api_key(script_hint="python3 skill/uwillberich/scripts/runtime_config.py set-em-key --stdin") def post_json(path: str, payload: dict, timeout: int = 30) -> dict: url = f"{MX_BASE_URL}/{path.lstrip('/')}" request = urllib.request.Request( url, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers={**DEFAULT_HEADERS, "apikey": get_mx_api_key()}, ) with urllib.request.urlopen(request, timeout=timeout) as response: return json.loads(response.read().decode("utf-8", errors="replace")) def unwrap_response(payload: dict) -> dict: data = payload.get("data") while isinstance(data, dict) and "data" in data: next_data = data.get("data") if next_data is None: break data = next_data return data if isinstance(data, dict) else {} def news_search(query: str, size: int | None = None) -> dict: payload = {"query": query} if size is not None: payload["size"] = size response = post_json("news-search", payload) data = unwrap_response(response) items = ((data.get("llmSearchResponse") or {}).get("data")) or [] return {"query": query, "items": items, "raw": response} def stock_screen(keyword: str, page_no: int = 1, page_size: int = 20) -> dict: payload = {"keyword": keyword, "pageNo": page_no, "pageSize": page_size} response = post_json("stock-screen", payload) data = unwrap_response(response) result = ((data.get("allResults") or {}).get("result")) or {} columns = result.get("columns") or [] rows = result.get("dataList") or [] return { "keyword": keyword, "title": data.get("title") or keyword, "response_code": data.get("responseCode"), "reflect_result": data.get("reflectResult"), "security_count": data.get("securityCount"), "conditions": data.get("responseConditionList") or [], "columns": columns, "rows": rows, "total": result.get("total") or len(rows), "raw": response, } def data_query(tool_query: str) -> dict: response = post_json("query", {"toolQuery": tool_query}) data = unwrap_response(response) result = data.get("searchDataResultDTO") or {} tables = result.get("dataTableDTOList") or [] entities = result.get("entityTagDTOList") or [] return { "tool_query": tool_query, "question_id": result.get("questionId"), "tables": tables, "entities": entities, "condition": result.get("condition") or {}, "raw": response, } def format_news_markdown(items: list[dict], limit: int = 5) -> str: lines = ["| Date | Source | Title | Type |", "| --- | --- | --- | --- |"] for item in items[:limit]: title = item.get("title") or "" url = item.get("jumpUrl") or "" linked_title = f"[{title}]({url})" if url else title lines.append( f"| {item.get('date', '')} | {item.get('source', '')} | {linked_title} | {item.get('informationType', '')} |" ) return "\n".join(lines) def csv_header(columns: list[dict]) -> list[str]: headers = [] for column in columns: title = column.get("title") or column.get("key") or "" date_msg = column.get("dateMsg") if date_msg: title = f"{title}({date_msg})" headers.append(title) return headers def csv_keys(columns: list[dict]) -> list[str]: return [column.get("key") or "" for column in columns] def write_stock_screen_csv(columns: list[dict], rows: list[dict], path: str) -> None: output_path = Path(path) output_path.parent.mkdir(parents=True, exist_ok=True) keys = csv_keys(columns) headers = csv_header(columns) with output_path.open("w", encoding="utf-8", newline="") as handle: writer = csv.writer(handle) writer.writerow(headers) for row in rows: writer.writerow([row.get(key, "") for key in keys]) def write_stock_screen_description(columns: list[dict], path: str) -> None: output_path = Path(path) output_path.parent.mkdir(parents=True, exist_ok=True) lines = ["# Stock Screen Columns", "", "| 标题 | 字段 key | 日期 | 单位 | 数据类型 |", "| --- | --- | --- | --- | --- |"] for column in columns: lines.append( f"| {column.get('title', '')} | {column.get('key', '')} | {column.get('dateMsg', '') or ''} | {column.get('unit', '') or ''} | {column.get('dataType', '') or ''} |" ) output_path.write_text("\n".join(lines) + "\n", encoding="utf-8") def format_stock_screen_markdown(columns: list[dict], rows: list[dict], limit: int = 10) -> str: keys = csv_keys(columns) header_map = dict(zip(keys, csv_header(columns))) preferred_matchers = [ "SERIAL", "SECURITY_CODE", "SECURITY_SHORT_NAME", "MARKET_SHORT_NAME", "NEWEST_PRICE", "CHG", "PCHG", "010000_RPT_F10_ORG_BASICINFO_BOARD_NAME_TOTAL_BOARD_NAME_TOTAL_", "010000_TOAL_MARKET_VALUE", "010000_CIRCULATION_MARKET_VALUE", ] top_keys: list[str] = [] for matcher in preferred_matchers: match = next((key for key in keys if key == matcher or key.startswith(matcher)), None) if match and match not in top_keys: top_keys.append(match) for key in keys: if key not in top_keys: top_keys.append(key) if len(top_keys) >= 8: break top_headers = [header_map[key] for key in top_keys] lines = ["| " + " | ".join(top_headers) + " |", "| " + " | ".join(["---"] * len(top_headers)) + " |"] for row in rows[:limit]: values = [str(row.get(key, "")) for key in top_keys] lines.append("| " + " | ".join(values) + " |") return "\n".join(lines) def extract_latest_metrics(table: dict) -> list[dict]: metrics: list[dict] = [] name_map = table.get("nameMap") or {} data_table = table.get("table") or {} dates = data_table.get("headName") or [] as_of = dates[0] if dates else "" indicator_order = table.get("indicatorOrder") or [] for key in indicator_order: if key == "headName": continue values = data_table.get(key) or [] if not values: continue metrics.append( { "entity": table.get("entityName", ""), "metric": name_map.get(key, key), "latest": values[0], "as_of": as_of, "title": table.get("title", ""), } ) return metrics def format_data_query_markdown(tables: list[dict], limit: int = 12) -> str: metrics: list[dict] = [] seen: set[tuple[str, str, str, str]] = set() for table in tables: for item in extract_latest_metrics(table): fingerprint = (item["entity"], item["metric"], str(item["latest"]), item["as_of"]) if fingerprint in seen: continue seen.add(fingerprint) metrics.append(item) lines = ["| Entity | Metric | Latest | As Of |", "| --- | --- | --- | --- |"] for item in metrics[:limit]: lines.append( f"| {item['entity']} | {item['metric']} | {item['latest']} | {item['as_of']} |" ) return "\n".join(lines) FILE:scripts/mx_toolkit.py #!/usr/bin/env python3 from __future__ import annotations import argparse import json import re from datetime import datetime from pathlib import Path from mx_api import ( data_query, format_data_query_markdown, format_news_markdown, format_stock_screen_markdown, news_search, stock_screen, write_stock_screen_csv, write_stock_screen_description, ) from runtime_config import get_output_dir, require_em_api_key ROOT = Path(__file__).resolve().parents[1] DEFAULT_PRESET_PATH = ROOT / "assets" / "mx_presets.json" def load_presets(path: str) -> dict: return json.loads(Path(path).read_text(encoding="utf-8")) def slugify(value: str) -> str: text = re.sub(r"[^a-zA-Z0-9\u4e00-\u9fff]+", "-", value.strip()).strip("-") return text or "step" def maybe_write_json(path: Path | None, payload: dict) -> None: if not path: return path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def maybe_write_text(path: Path | None, content: str) -> None: if not path: return path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content.rstrip() + "\n", encoding="utf-8") def render_news_markdown(query: str, result: dict, limit: int) -> str: lines = [f"# MX News Search", "", f"Query: `{query}`", ""] lines.append(format_news_markdown(result["items"], limit=limit)) return "\n".join(lines).rstrip() + "\n" def render_stock_screen_markdown(result: dict, limit: int) -> str: lines = [f"# MX Stock Screen", "", f"Keyword: `{result['keyword']}`", ""] lines.append(f"- response_code: `{result['response_code']}`") lines.append(f"- reflect_result: `{result['reflect_result']}`") lines.append(f"- security_count: `{result['security_count']}`") for item in result["conditions"]: lines.append(f"- condition: {item.get('describe', '')} -> {item.get('stockCount', '')}") lines.append("") lines.append(format_stock_screen_markdown(result["columns"], result["rows"], limit=limit)) return "\n".join(lines).rstrip() + "\n" def render_data_query_markdown(result: dict, limit: int) -> str: lines = [f"# MX Data Query", "", f"Tool Query: `{result['tool_query']}`", ""] lines.append(f"- question_id: `{result['question_id'] or 'n/a'}`") lines.append(f"- tables: `{len(result['tables'])}`") lines.append(f"- entities: `{len(result['entities'])}`") lines.append("") lines.append(format_data_query_markdown(result["tables"], limit=limit)) return "\n".join(lines).rstrip() + "\n" def save_single_run_outputs(command: str, output_dir: str | None) -> Path | None: if not output_dir: return None target = Path(output_dir).expanduser() target.mkdir(parents=True, exist_ok=True) return target def run_news_search(args: argparse.Namespace) -> int: result = news_search(args.query, size=args.size) markdown = render_news_markdown(args.query, result, args.limit) output_dir = save_single_run_outputs("news-search", args.output_dir) maybe_write_json(output_dir / "raw.json" if output_dir else None, result["raw"]) maybe_write_text(output_dir / "report.md" if output_dir else None, markdown) if args.format == "json": print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 print(markdown) if output_dir: print(f"Saved: {output_dir}") return 0 def run_stock_screen(args: argparse.Namespace) -> int: result = stock_screen(args.keyword, page_no=args.page_no, page_size=args.page_size) markdown = render_stock_screen_markdown(result, args.limit) output_dir = save_single_run_outputs("stock-screen", args.output_dir) maybe_write_json(output_dir / "raw.json" if output_dir else None, result["raw"]) maybe_write_text(output_dir / "report.md" if output_dir else None, markdown) csv_out = Path(args.csv_out).expanduser() if args.csv_out else (output_dir / "screen.csv" if output_dir else None) desc_out = Path(args.desc_out).expanduser() if args.desc_out else (output_dir / "columns.md" if output_dir else None) if csv_out: write_stock_screen_csv(result["columns"], result["rows"], str(csv_out)) if desc_out: write_stock_screen_description(result["columns"], str(desc_out)) if args.format == "json": print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 print(markdown) if csv_out: print(f"CSV: {csv_out}") if desc_out: print(f"Columns: {desc_out}") if output_dir: print(f"Saved: {output_dir}") return 0 def run_query(args: argparse.Namespace) -> int: result = data_query(args.tool_query) markdown = render_data_query_markdown(result, args.limit) output_dir = save_single_run_outputs("query", args.output_dir) maybe_write_json(output_dir / "raw.json" if output_dir else None, result["raw"]) maybe_write_text(output_dir / "report.md" if output_dir else None, markdown) if args.format == "json": print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 print(markdown) if output_dir: print(f"Saved: {output_dir}") return 0 def render_preset_step(step: dict, result: dict) -> str: tool = step["tool"] if tool == "news-search": return render_news_markdown(step["query"], result, step.get("limit", 5)) if tool == "stock-screen": return render_stock_screen_markdown(result, step.get("limit", 10)) if tool == "query": return render_data_query_markdown(result, step.get("limit", 12)) raise ValueError(f"unsupported tool: {tool}") def execute_preset_step(step: dict, output_dir: Path) -> dict: tool = step["tool"] slug = step.get("slug") or slugify(step.get("query") or step.get("keyword") or step.get("tool_query") or tool) step_dir = output_dir / slug step_dir.mkdir(parents=True, exist_ok=True) if tool == "news-search": result = news_search(step["query"], size=step.get("size")) markdown = render_news_markdown(step["query"], result, step.get("limit", 5)) maybe_write_json(step_dir / "raw.json", result["raw"]) maybe_write_text(step_dir / "report.md", markdown) return {"tool": tool, "slug": slug, "markdown": markdown, "saved_dir": str(step_dir)} if tool == "stock-screen": result = stock_screen(step["keyword"], page_no=step.get("page_no", 1), page_size=step.get("page_size", 20)) markdown = render_stock_screen_markdown(result, step.get("limit", 10)) maybe_write_json(step_dir / "raw.json", result["raw"]) maybe_write_text(step_dir / "report.md", markdown) write_stock_screen_csv(result["columns"], result["rows"], str(step_dir / "screen.csv")) write_stock_screen_description(result["columns"], str(step_dir / "columns.md")) return { "tool": tool, "slug": slug, "markdown": markdown, "saved_dir": str(step_dir), "security_count": result["security_count"], } if tool == "query": result = data_query(step["tool_query"]) markdown = render_data_query_markdown(result, step.get("limit", 12)) maybe_write_json(step_dir / "raw.json", result["raw"]) maybe_write_text(step_dir / "report.md", markdown) return {"tool": tool, "slug": slug, "markdown": markdown, "saved_dir": str(step_dir)} raise ValueError(f"unsupported tool: {tool}") def run_list_presets(args: argparse.Namespace) -> int: presets = load_presets(args.preset_path) print("# MX Presets\n") for name, config in presets.items(): print(f"- `{name}`: {config.get('description', '')}") return 0 def run_preset(args: argparse.Namespace) -> int: presets = load_presets(args.preset_path) if args.name not in presets: available = ", ".join(sorted(presets)) raise SystemExit(f"unknown preset: {args.name}. available: {available}") preset = presets[args.name] base_dir = Path(args.output_dir).expanduser() if args.output_dir else get_output_dir("mx-presets") timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_dir = base_dir / f"{args.name}_{timestamp}" output_dir.mkdir(parents=True, exist_ok=True) sections = [f"# MX Preset Run", "", f"Preset: `{args.name}`", ""] if preset.get("description"): sections.append(preset["description"]) sections.append("") for index, step in enumerate(preset.get("steps", []), start=1): result = execute_preset_step(step, output_dir) sections.append(f"## Step {index}: {result['slug']}") sections.append(f"- tool: `{result['tool']}`") sections.append(f"- saved_dir: `{result['saved_dir']}`") sections.append("") sections.append(result["markdown"].strip()) sections.append("") report = "\n".join(sections).rstrip() + "\n" maybe_write_text(output_dir / "preset_report.md", report) if args.format == "json": print( json.dumps( { "preset": args.name, "description": preset.get("description", ""), "output_dir": str(output_dir), "steps": preset.get("steps", []), }, ensure_ascii=False, indent=2, ) ) return 0 print(report) print(f"Preset output: {output_dir}") return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Use the Meixiang / Eastmoney APIs with the local EM_API_KEY.") subparsers = parser.add_subparsers(dest="command", required=True) presets_parser = subparsers.add_parser("list-presets", help="List the preset MX workflows.") presets_parser.add_argument("--preset-path", default=str(DEFAULT_PRESET_PATH), help="Preset config JSON path.") presets_parser.set_defaults(func=run_list_presets) preset_parser = subparsers.add_parser("preset", help="Run a preset MX workflow and save all artifacts.") preset_parser.add_argument("--name", required=True, help="Preset name.") preset_parser.add_argument("--preset-path", default=str(DEFAULT_PRESET_PATH), help="Preset config JSON path.") preset_parser.add_argument("--output-dir", help="Optional base output directory.") preset_parser.add_argument("--format", choices=["markdown", "json"], default="markdown") preset_parser.set_defaults(func=run_preset) news_parser = subparsers.add_parser("news-search", help="Run a real MX financial news search.") news_parser.add_argument("--query", required=True, help="Natural-language financial news query.") news_parser.add_argument("--size", type=int, default=8, help="Requested result size.") news_parser.add_argument("--limit", type=int, default=5, help="Rendered result limit.") news_parser.add_argument("--format", choices=["markdown", "json"], default="markdown") news_parser.add_argument("--output-dir", help="Optional directory to save raw JSON and markdown report.") news_parser.set_defaults(func=run_news_search) screen_parser = subparsers.add_parser("stock-screen", help="Run a real MX stock screen and optionally export CSV.") screen_parser.add_argument("--keyword", required=True, help="Natural-language stock-screen query.") screen_parser.add_argument("--page-no", type=int, default=1, help="Page number.") screen_parser.add_argument("--page-size", type=int, default=20, help="Page size.") screen_parser.add_argument("--limit", type=int, default=10, help="Rendered row limit.") screen_parser.add_argument("--format", choices=["markdown", "json"], default="markdown") screen_parser.add_argument("--output-dir", help="Optional directory to save raw JSON, report, CSV, and columns.") screen_parser.add_argument("--csv-out", help="Optional path to save the full result CSV.") screen_parser.add_argument("--desc-out", help="Optional path to save a columns description markdown file.") screen_parser.set_defaults(func=run_stock_screen) query_parser = subparsers.add_parser("query", help="Run a real MX structured data query.") query_parser.add_argument("--tool-query", required=True, help="Natural-language data query.") query_parser.add_argument("--limit", type=int, default=12, help="Rendered metric limit.") query_parser.add_argument("--format", choices=["markdown", "json"], default="markdown") query_parser.add_argument("--output-dir", help="Optional directory to save raw JSON and markdown report.") query_parser.set_defaults(func=run_query) return parser def main() -> int: require_em_api_key(script_hint="python3 skill/uwillberich/scripts/runtime_config.py set-em-key --stdin") parser = build_parser() args = parser.parse_args() return args.func(args) if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/news_iterator.py #!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import html import json import re import sqlite3 import sys import time import urllib.request import xml.etree.ElementTree as ET from collections import Counter from dataclasses import dataclass from datetime import UTC, datetime, timedelta from email.utils import parsedate_to_datetime from pathlib import Path from industry_chain import enrich_event_payload_with_chain_focus from runtime_config import load_runtime_env, require_em_api_key ROOT = Path(__file__).resolve().parents[1] DEFAULT_CONFIG = ROOT / "assets" / "news_iterator_config.json" DEFAULT_WATCHLIST = ROOT / "assets" / "default_watchlists.json" DEFAULT_STATE_DIR = Path.home() / ".uwillberich" / "news-iterator" DEFAULT_DB = DEFAULT_STATE_DIR / "news_iterator.sqlite3" DEFAULT_MARKDOWN = DEFAULT_STATE_DIR / "latest_alerts.md" DEFAULT_JSONL = DEFAULT_STATE_DIR / "alerts.jsonl" DEFAULT_EVENT_WATCHLIST = DEFAULT_STATE_DIR / "event_watchlists.json" DEFAULT_HEADERS = {"User-Agent": "Mozilla/5.0"} EVENT_CATEGORY_ORDER = ["huge_conflict", "huge_future", "huge_name_release"] CATEGORY_LABELS = { "huge_conflict": "巨大冲突", "huge_future": "巨大前景", "huge_name_release": "巨头名人", } SIGNAL_LABELS = {"high": "高", "medium": "中", "low": "低"} KEYWORD_LABELS = { "war": "战争", "oil": "原油", "energy": "能源", "chips": "芯片", "chip": "芯片", "robots": "机器人", "robot": "机器人", "launch": "发布", "launches": "发布", "announces": "宣布", "announce": "宣布", "unveils": "亮相", "unveil": "亮相", "data center": "数据中心", } load_runtime_env() require_em_api_key(script_hint="python3 skill/uwillberich/scripts/runtime_config.py set-em-key --stdin") @dataclass class FeedItem: item_key: str feed_key: str feed_label: str source: str title: str link: str summary: str published_at: str def load_config(path: str) -> dict: return json.loads(Path(path).read_text(encoding="utf-8")) def ensure_state_dir(path: Path) -> None: path.mkdir(parents=True, exist_ok=True) def open_db(path: Path) -> sqlite3.Connection: ensure_state_dir(path.parent) conn = sqlite3.connect(path) conn.execute( """ CREATE TABLE IF NOT EXISTS items ( item_key TEXT PRIMARY KEY, feed_key TEXT NOT NULL, feed_label TEXT NOT NULL, source TEXT, title TEXT NOT NULL, link TEXT NOT NULL, summary TEXT, published_at TEXT, inserted_at TEXT NOT NULL ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS alerts ( alert_id INTEGER PRIMARY KEY AUTOINCREMENT, item_key TEXT NOT NULL, category TEXT NOT NULL, score INTEGER NOT NULL, signal TEXT NOT NULL, impacted_watchlists_json TEXT NOT NULL, watchlist_scores_json TEXT NOT NULL DEFAULT '{}', matched_entities_json TEXT NOT NULL, matched_keywords_json TEXT NOT NULL, created_at TEXT NOT NULL, UNIQUE(item_key, category) ) """ ) columns = {row[1] for row in conn.execute("PRAGMA table_info(alerts)").fetchall()} if "watchlist_scores_json" not in columns: conn.execute("ALTER TABLE alerts ADD COLUMN watchlist_scores_json TEXT NOT NULL DEFAULT '{}'") return conn def normalize_text(value: str) -> str: cleaned = value or "" cleaned = re.sub(r"<[^>]+>", " ", cleaned) cleaned = html.unescape(cleaned) return re.sub(r"\s+", " ", cleaned).strip() def normalize_match_text(value: str) -> str: cleaned = normalize_text(value) cleaned = re.sub(r"https?://\S+", " ", cleaned) cleaned = re.sub(r"\bnews\.google\.com\b", " ", cleaned, flags=re.IGNORECASE) return cleaned.lower() def category_display_name(category: str) -> str: return CATEGORY_LABELS.get(category, category) def signal_display_name(signal: str) -> str: return SIGNAL_LABELS.get(signal, signal) def keyword_display_name(keyword: str) -> str: return KEYWORD_LABELS.get(keyword, keyword) def format_keyword_list(keywords: list[str]) -> str: if not keywords: return "n/a" return ", ".join(keyword_display_name(keyword) for keyword in keywords) def term_pattern(term: str) -> re.Pattern[str]: escaped = re.escape(normalize_match_text(term)) return re.compile(rf"(? bool: return bool(term_pattern(term).search(normalize_match_text(text))) def parse_datetime(raw: str) -> str: if not raw: return "" try: parsed = parsedate_to_datetime(raw) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=UTC) return parsed.astimezone(UTC).isoformat() except Exception: return raw def fetch_url(url: str) -> bytes: request = urllib.request.Request(url, headers=DEFAULT_HEADERS) with urllib.request.urlopen(request, timeout=20) as response: return response.read() def build_item_key(feed_key: str, guid: str, link: str, title: str) -> str: base = guid or link or title return hashlib.sha256(f"{feed_key}|{base}".encode("utf-8")).hexdigest() def parse_feed(feed: dict) -> list[FeedItem]: payload = fetch_url(feed["url"]) root = ET.fromstring(payload) items: list[FeedItem] = [] channel = root.find("channel") if channel is not None: for item in channel.findall("item"): title = normalize_text(item.findtext("title")) link = normalize_text(item.findtext("link")) summary = normalize_text(item.findtext("description")) source = normalize_text(item.findtext("source")) or feed["label"] guid = normalize_text(item.findtext("guid")) published = parse_datetime(normalize_text(item.findtext("pubDate"))) items.append( FeedItem( item_key=build_item_key(feed["key"], guid, link, title), feed_key=feed["key"], feed_label=feed["label"], source=source, title=title, link=link, summary=summary, published_at=published, ) ) return items ns = {"atom": "http://www.w3.org/2005/Atom"} for entry in root.findall("atom:entry", ns): title = normalize_text(entry.findtext("atom:title", default="", namespaces=ns)) link_el = entry.find("atom:link", ns) link = normalize_text(link_el.attrib.get("href", "")) if link_el is not None else "" summary = normalize_text(entry.findtext("atom:summary", default="", namespaces=ns)) source = feed["label"] guid = normalize_text(entry.findtext("atom:id", default="", namespaces=ns)) published = parse_datetime( normalize_text(entry.findtext("atom:updated", default="", namespaces=ns)) ) items.append( FeedItem( item_key=build_item_key(feed["key"], guid, link, title), feed_key=feed["key"], feed_label=feed["label"], source=source, title=title, link=link, summary=summary, published_at=published, ) ) return items def match_terms(text: str, terms: list[str]) -> list[str]: return sorted({term for term in terms if text_contains_term(text, term)}) def bump_watchlist_scores(scores: dict[str, int], groups: list[str], points: int) -> None: for group in groups: scores[group] = scores.get(group, 0) + points def derive_watchlist_scores( text: str, matched_entities: list[str], config: dict, categories: list[str], ) -> dict[str, int]: watchlist_scores: dict[str, int] = {} for entity in matched_entities: bump_watchlist_scores( watchlist_scores, config.get("entity_watchlists", {}).get(entity.lower(), []), points=2, ) for keyword, groups in config.get("keyword_watchlists", {}).items(): if text_contains_term(text, keyword): bump_watchlist_scores(watchlist_scores, groups, points=2) if "huge_future" in categories: bump_watchlist_scores( watchlist_scores, [ "cross_cycle_anchor12", "cross_cycle_ai_hardware", "cross_cycle_semis", "cross_cycle_software_platforms", ], points=1, ) if "huge_name_release" in categories: bump_watchlist_scores(watchlist_scores, ["cross_cycle_anchor12"], points=1) if "huge_conflict" in categories: bump_watchlist_scores( watchlist_scores, [ "war_shock_core12", "defensive_gauge", ] , points=1, ) bump_watchlist_scores(watchlist_scores, ["war_benefit_oil_coal"], points=1) bump_watchlist_scores(watchlist_scores, ["war_headwind_compute_power"], points=1) return watchlist_scores def score_to_signal(score: int) -> str: if score >= 10: return "high" if score >= 6: return "medium" return "low" def classify_item(item: FeedItem, config: dict) -> list[dict]: title_text = item.title.strip() text = f"{item.title} {item.summary}".strip() matched_entities = match_terms(title_text, config.get("big_name_entities", [])) matched_conflict_entities = match_terms(text, config.get("conflict_entities", [])) matched_future = match_terms(text, config.get("future_keywords", [])) matched_release = match_terms(title_text, config.get("release_verbs", [])) matched_conflict = match_terms(text, config.get("conflict_keywords", [])) matched_energy = match_terms(text, config.get("energy_keywords", [])) matched_compute_power = match_terms(text, config.get("compute_power_keywords", [])) alerts: list[dict] = [] if matched_future and not matched_conflict and not matched_conflict_entities: score = len(matched_future) * 2 + (2 if matched_entities else 0) categories = ["huge_future"] watchlist_scores = derive_watchlist_scores(text, matched_entities, config, categories) alerts.append( { "category": "huge_future", "score": score, "signal": score_to_signal(score), "matched_entities": matched_entities, "matched_keywords": matched_future, "impacted_watchlists": sorted( watchlist_scores, key=lambda group: (-watchlist_scores[group], group), ), "watchlist_scores": watchlist_scores, } ) if matched_entities and matched_release: score = len(matched_entities) * 3 + len(matched_release) * 2 categories = ["huge_name_release"] watchlist_scores = derive_watchlist_scores(text, matched_entities, config, categories) alerts.append( { "category": "huge_name_release", "score": score, "signal": score_to_signal(score), "matched_entities": matched_entities, "matched_keywords": matched_release, "impacted_watchlists": sorted( watchlist_scores, key=lambda group: (-watchlist_scores[group], group), ), "watchlist_scores": watchlist_scores, } ) if matched_conflict or matched_conflict_entities: score = len(matched_conflict) * 3 + len(matched_conflict_entities) * 3 if matched_energy: score += 2 if matched_compute_power: score += 1 categories = ["huge_conflict"] all_entities = sorted(set(matched_conflict_entities + matched_entities)) watchlist_scores = derive_watchlist_scores(text, all_entities, config, categories) alerts.append( { "category": "huge_conflict", "score": score, "signal": score_to_signal(score), "matched_entities": all_entities, "matched_keywords": sorted(set(matched_conflict + matched_energy + matched_compute_power)), "impacted_watchlists": sorted( watchlist_scores, key=lambda group: (-watchlist_scores[group], group), ), "watchlist_scores": watchlist_scores, } ) return [alert for alert in alerts if alert["score"] >= 4] def item_exists(conn: sqlite3.Connection, item_key: str) -> bool: row = conn.execute("SELECT 1 FROM items WHERE item_key = ?", (item_key,)).fetchone() return row is not None def insert_item(conn: sqlite3.Connection, item: FeedItem) -> None: conn.execute( """ INSERT OR IGNORE INTO items ( item_key, feed_key, feed_label, source, title, link, summary, published_at, inserted_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( item.item_key, item.feed_key, item.feed_label, item.source, item.title, item.link, item.summary, item.published_at, datetime.now(UTC).isoformat(), ), ) def insert_alert(conn: sqlite3.Connection, item: FeedItem, alert: dict) -> bool: cursor = conn.execute( """ INSERT OR IGNORE INTO alerts ( item_key, category, score, signal, impacted_watchlists_json, matched_entities_json, matched_keywords_json, watchlist_scores_json, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( item.item_key, alert["category"], alert["score"], alert["signal"], json.dumps(alert["impacted_watchlists"], ensure_ascii=False), json.dumps(alert["matched_entities"], ensure_ascii=False), json.dumps(alert["matched_keywords"], ensure_ascii=False), json.dumps(alert.get("watchlist_scores", {}), ensure_ascii=False), datetime.now(UTC).isoformat(), ), ) return cursor.rowcount > 0 def fetch_and_classify(conn: sqlite3.Connection, config: dict) -> list[dict]: new_alerts: list[dict] = [] for feed in config.get("feeds", []): try: items = parse_feed(feed) except Exception as exc: new_alerts.append( { "system_error": True, "feed_key": feed["key"], "feed_label": feed["label"], "error": str(exc), } ) continue for item in items: is_new_item = not item_exists(conn, item.item_key) if is_new_item: insert_item(conn, item) alerts = classify_item(item, config) for alert in alerts: if insert_alert(conn, item, alert): row = {"item": item, "alert": alert} new_alerts.append(row) conn.commit() return new_alerts def row_to_markdown(row: dict) -> str: item: FeedItem = row["item"] alert = row["alert"] return ( f"- [{item.title}]({item.link})\n" f" source: {item.source}\n" f" category: `{alert['category']}` | signal: `{alert['signal']}` | score: `{alert['score']}`\n" f" watchlists: {', '.join(alert['impacted_watchlists']) or 'n/a'}\n" f" entities: {', '.join(alert['matched_entities']) or 'n/a'}\n" f" keywords: {', '.join(alert['matched_keywords']) or 'n/a'}" ) def append_jsonl(new_alerts: list[dict], jsonl_path: Path) -> None: ensure_state_dir(jsonl_path.parent) json_lines: list[str] = [] for row in new_alerts: if row.get("system_error"): json_lines.append(json.dumps(row, ensure_ascii=False)) else: item = row["item"] alert = row["alert"] json_lines.append( json.dumps( { "item_key": item.item_key, "title": item.title, "link": item.link, "source": item.source, "published_at": item.published_at, "category": alert["category"], "score": alert["score"], "signal": alert["signal"], "impacted_watchlists": alert["impacted_watchlists"], "watchlist_scores": alert.get("watchlist_scores", {}), "matched_entities": alert["matched_entities"], "matched_keywords": alert["matched_keywords"], }, ensure_ascii=False, ) ) with jsonl_path.open("a", encoding="utf-8") as handle: for line in json_lines: handle.write(line + "\n") def fetch_recent_alerts(conn: sqlite3.Connection, hours: int) -> list[dict]: cutoff = (datetime.now(UTC) - timedelta(hours=hours)).isoformat() rows = conn.execute( """ SELECT a.category, a.score, a.signal, a.impacted_watchlists_json, a.watchlist_scores_json, a.matched_entities_json, a.matched_keywords_json, i.title, i.link, i.source, i.published_at FROM alerts a JOIN items i ON i.item_key = a.item_key WHERE a.created_at >= ? ORDER BY a.score DESC, a.created_at DESC """, (cutoff,), ).fetchall() result = [] for row in rows: result.append( { "category": row[0], "score": row[1], "signal": row[2], "watchlists": json.loads(row[3]), "watchlist_scores": json.loads(row[4] or "{}"), "entities": json.loads(row[5]), "keywords": json.loads(row[6]), "title": row[7], "link": row[8], "source": row[9], "published_at": row[10], } ) return result def top_alerts_by_category(alerts: list[dict], limit: int = 10) -> dict[str, list[dict]]: grouped: dict[str, list[dict]] = {} for category in EVENT_CATEGORY_ORDER: ranked = sorted( [alert for alert in alerts if alert["category"] == category], key=lambda item: ( item["score"], item.get("published_at") or "", item.get("title") or item.get("headline") or "", ), reverse=True, ) if ranked: deduped: list[dict] = [] seen_links: set[str] = set() for item in ranked: link = item.get("link") or item.get("title") if link in seen_links: continue seen_links.add(link) deduped.append(item) if len(deduped) >= limit: break grouped[category] = deduped return grouped def render_report(alerts: list[dict], hours: int) -> str: lines = [f"# News Iterator Report", f"", f"Window: last {hours} hours"] if not alerts: lines.append("\nNo alerts in the selected window.") return "\n".join(lines) + "\n" summary = summarize_alert_categories(alerts) if summary: lines.append("") lines.append("## Event Summary") lines.append("") lines.append("| 类别 | 条数 | 总分 | 高频关键词 |") lines.append("| --- | ---: | ---: | --- |") for item in summary: lines.append( f"| {category_display_name(item['category'])} | {item['alert_count']} | {item['total_score']} | {format_keyword_list(item.get('top_keywords', []))} |" ) grouped = top_alerts_by_category(alerts, limit=10) for category in EVENT_CATEGORY_ORDER: items = grouped.get(category, []) if not items: continue lines.append(f"\n## {category_display_name(category)} Top 10 信息源") for index, alert in enumerate(items, start=1): lines.append(f"{index}. [{alert['title']}]({alert['link']})") lines.append( f" - 来源: {alert['source']} | 信号: `{signal_display_name(alert['signal'])}` | 分值: `{alert['score']}`" ) lines.append(f" - 实体: {', '.join(alert['entities']) or 'n/a'}") lines.append(f" - 关键词: {format_keyword_list(alert['keywords'])}") return "\n".join(lines) + "\n" def render_system_errors(rows: list[dict]) -> str: if not rows: return "" lines = ["\n## system_error"] for row in rows: lines.append(f"- feed: `{row['feed_key']}` ({row['feed_label']}) | error: {row['error']}") return "\n".join(lines) + "\n" def load_base_watchlists(path: str) -> dict: return json.loads(Path(path).read_text(encoding="utf-8")) def merge_item_details(existing: dict, incoming: dict) -> dict: merged = dict(existing) for key, value in incoming.items(): if not merged.get(key) and value: merged[key] = value return merged def build_base_item_index(base_watchlists: dict) -> tuple[dict[str, dict], dict[str, list[dict]]]: symbol_index: dict[str, dict] = {} group_index: dict[str, list[dict]] = {} for group, items in base_watchlists.items(): group_index[group] = [] for item in items: symbol = item["symbol"] if symbol in symbol_index: symbol_index[symbol] = merge_item_details(symbol_index[symbol], item) else: symbol_index[symbol] = dict(item) group_index[group].append(symbol_index[symbol]) return symbol_index, group_index def shorten_driver(category: str, keywords: Counter[str], entities: Counter[str]) -> str: top_terms = [term for term, _ in keywords.most_common(3)] top_entities = [entity for entity, _ in entities.most_common(2)] parts = [category] if top_terms: parts.append("/".join(top_terms)) if top_entities: parts.append(",".join(top_entities)) return " | ".join(parts) def build_event_item(category: str, item: dict, stats: dict) -> dict: category_label = category.replace("event_focus_", "") if category.startswith("event_focus_") else category role = item.get("role", "") strong_signal = item.get("strong_signal") or "消息驱动仍在扩散时，优先看它能否领涨并放量。" weak_signal = item.get("weak_signal") or "消息很多但股价不跟，说明事件交易开始钝化。" return { "symbol": item["symbol"], "name": item.get("name", ""), "role": role, "event_score": stats["event_score"], "trigger_count": stats["trigger_count"], "event_driver": shorten_driver(category_label, stats["keywords"], stats["entities"]), "source_groups": sorted(stats["source_groups"]), "trigger_categories": sorted(stats["categories"]), "strong_signal": strong_signal, "weak_signal": weak_signal, } def aggregate_alerts_into_pool( alerts: list[dict], group_index: dict[str, list[dict]], symbol_index: dict[str, dict], allowed_groups: set[str] | None = None, ) -> dict[str, dict]: symbol_stats: dict[str, dict] = {} for alert in alerts: symbol_weights: dict[str, int] = {} symbol_source_groups: dict[str, set[str]] = {} watchlist_scores = alert.get("watchlist_scores") or {group: 1 for group in alert.get("watchlists", [])} for group, group_weight in watchlist_scores.items(): if group not in group_index: continue if allowed_groups is not None and group not in allowed_groups: continue for item in group_index[group]: symbol = item["symbol"] symbol_weights[symbol] = max(symbol_weights.get(symbol, 0), group_weight) symbol_source_groups.setdefault(symbol, set()).add(group) if not symbol_weights: continue for symbol, weight in symbol_weights.items(): if symbol not in symbol_stats: symbol_stats[symbol] = { "event_score": 0, "trigger_count": 0, "keywords": Counter(), "entities": Counter(), "categories": set(), "source_groups": set(), } stats = symbol_stats[symbol] stats["event_score"] += alert["score"] * weight stats["trigger_count"] += 1 stats["keywords"].update(alert.get("keywords", [])) stats["entities"].update(alert.get("entities", [])) stats["categories"].add(alert["category"]) stats["source_groups"].update(symbol_source_groups.get(symbol, set())) return symbol_stats def rank_pool_items(symbol_stats: dict[str, dict], symbol_index: dict[str, dict], limit: int, category: str) -> list[dict]: ranked = sorted( symbol_stats.items(), key=lambda pair: (-pair[1]["event_score"], -pair[1]["trigger_count"], pair[0]), ) items: list[dict] = [] for symbol, stats in ranked[:limit]: if symbol not in symbol_index: continue items.append(build_event_item(category, symbol_index[symbol], stats)) return items def summarize_alert_categories(alerts: list[dict]) -> list[dict]: category_map: dict[str, dict] = {} for alert in alerts: bucket = category_map.setdefault( alert["category"], {"category": alert["category"], "alert_count": 0, "total_score": 0, "top_keywords": Counter()}, ) bucket["alert_count"] += 1 bucket["total_score"] += alert["score"] bucket["top_keywords"].update(alert.get("keywords", [])) summary = [] for bucket in category_map.values(): summary.append( { "category": bucket["category"], "alert_count": bucket["alert_count"], "total_score": bucket["total_score"], "top_keywords": [term for term, _ in bucket["top_keywords"].most_common(3)], } ) return sorted(summary, key=lambda item: (-item["total_score"], -item["alert_count"], item["category"])) def build_event_watchlists_payload(alerts: list[dict], base_watchlists: dict, hours: int) -> dict: symbol_index, group_index = build_base_item_index(base_watchlists) groups: dict[str, list[dict]] = {} all_stats = aggregate_alerts_into_pool(alerts, group_index, symbol_index) groups["event_focus_core"] = rank_pool_items(all_stats, symbol_index, limit=12, category="event_focus_core") category_summary = summarize_alert_categories(alerts) for category in EVENT_CATEGORY_ORDER: category_alerts = [alert for alert in alerts if alert["category"] == category] if not category_alerts: continue stats = aggregate_alerts_into_pool(category_alerts, group_index, symbol_index) groups[f"event_focus_{category}"] = rank_pool_items( stats, symbol_index, limit=10, category=f"event_focus_{category}", ) conflict_alerts = [alert for alert in alerts if alert["category"] == "huge_conflict"] if conflict_alerts: benefit_stats = aggregate_alerts_into_pool( conflict_alerts, group_index, symbol_index, allowed_groups={"war_benefit_oil_coal"}, ) headwind_stats = aggregate_alerts_into_pool( conflict_alerts, group_index, symbol_index, allowed_groups={"war_headwind_compute_power"}, ) defensive_stats = aggregate_alerts_into_pool( conflict_alerts, group_index, symbol_index, allowed_groups={"defensive_gauge"}, ) groups["event_focus_huge_conflict_benefit"] = rank_pool_items( benefit_stats, symbol_index, limit=8, category="event_focus_huge_conflict_benefit", ) groups["event_focus_huge_conflict_headwind"] = rank_pool_items( headwind_stats, symbol_index, limit=8, category="event_focus_huge_conflict_headwind", ) groups["event_focus_huge_conflict_defensive"] = rank_pool_items( defensive_stats, symbol_index, limit=6, category="event_focus_huge_conflict_defensive", ) default_report_groups: list[str] = [] for item in category_summary[:2]: if item["category"] == "huge_conflict": for group_name in [ "event_focus_huge_conflict_benefit", "event_focus_huge_conflict_headwind", "event_focus_huge_conflict_defensive", ]: if group_name in groups: default_report_groups.append(group_name) continue group_name = f"event_focus_{item['category']}" if group_name in groups: default_report_groups.append(group_name) if not default_report_groups and groups.get("event_focus_core"): default_report_groups.append("event_focus_core") return { "generated_at": datetime.now(UTC).isoformat(), "lookback_hours": hours, "summary": category_summary, "top_alerts": top_alerts_by_category(alerts, limit=10), "default_report_groups": list(dict.fromkeys(default_report_groups)), "groups": {name: items for name, items in groups.items() if items}, } def write_event_watchlists(payload: dict, path: Path) -> None: ensure_state_dir(path.parent) path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def render_event_watchlists(payload: dict) -> str: groups = payload.get("groups", {}) if not groups: return "" lines = ["\n## Event Pools"] for group_name in payload.get("default_report_groups", []): items = groups.get(group_name, []) if not items: continue lines.append(f"\n### {group_name}") for item in items[:6]: lines.append( f"- {item['name']} `{item['symbol'][2:]}` | score `{item['event_score']}` | triggers `{item['trigger_count']}` | {item['event_driver']}" ) chain_summary = payload.get("chain_summary", []) if chain_summary: lines.append("\n## Industry Chain Focus") lines.append("") lines.append("| Theme | Score | Group | Reasons |") lines.append("| --- | ---: | --- | --- |") for item in chain_summary: reasons = " / ".join(item.get("reasons", [])[:3]) or "n/a" lines.append(f"| {item['theme']} | {item['score']} | {item['group']} | {reasons} |") for error in payload.get("chain_errors", []): lines.append(f"- chain_error: `{error['theme']}` | {error['error']}") return "\n".join(lines) + "\n" def run_poll(args: argparse.Namespace) -> int: config = load_config(args.config) conn = open_db(Path(args.db_path)) try: new_alerts = fetch_and_classify(conn, config) append_jsonl(new_alerts, Path(args.jsonl_path)) recent_alerts = fetch_recent_alerts(conn, args.report_hours) event_payload = build_event_watchlists_payload( recent_alerts, load_base_watchlists(args.watchlist_path), args.report_hours, ) event_payload = enrich_event_payload_with_chain_focus( event_payload, load_base_watchlists(args.watchlist_path), ) write_event_watchlists(event_payload, Path(args.event_watchlist_path)) markdown = render_report(recent_alerts, args.report_hours) markdown += render_event_watchlists(event_payload) system_errors = [row for row in new_alerts if row.get("system_error")] markdown += render_system_errors(system_errors) Path(args.markdown_path).write_text(markdown, encoding="utf-8") if args.format == "json": serializable = [] for row in new_alerts: if row.get("system_error"): serializable.append(row) continue item: FeedItem = row["item"] serializable.append( { "title": item.title, "link": item.link, "source": item.source, "published_at": item.published_at, **row["alert"], } ) print(json.dumps(serializable, ensure_ascii=False, indent=2)) else: print(markdown) return 0 finally: conn.close() def run_loop(args: argparse.Namespace) -> int: interval = max(args.interval_seconds, 30) while True: run_poll(args) time.sleep(interval) def run_report(args: argparse.Namespace) -> int: conn = open_db(Path(args.db_path)) try: alerts = fetch_recent_alerts(conn, args.hours) report = render_report(alerts, args.hours) event_payload = build_event_watchlists_payload( alerts, load_base_watchlists(args.watchlist_path), args.hours, ) event_payload = enrich_event_payload_with_chain_focus( event_payload, load_base_watchlists(args.watchlist_path), ) report += render_event_watchlists(event_payload) if args.event_watchlist_path: write_event_watchlists(event_payload, Path(args.event_watchlist_path)) print(report) return 0 finally: conn.close() def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Persistent RSS-based news iterator for A-share idea intake.") parser.add_argument("--config", default=str(DEFAULT_CONFIG), help="Path to news iterator config JSON.") parser.add_argument("--state-dir", default=str(DEFAULT_STATE_DIR), help="State directory for reports and DB.") parser.add_argument( "--watchlist-path", default=str(DEFAULT_WATCHLIST), help="Base watchlist JSON used to build dynamic event-driven stock pools.", ) subparsers = parser.add_subparsers(dest="command", required=True) def add_common_io(subparser: argparse.ArgumentParser) -> None: subparser.add_argument( "--db-path", default=str(DEFAULT_DB), help="SQLite database path. Defaults under the state directory.", ) subparser.add_argument( "--markdown-path", default=str(DEFAULT_MARKDOWN), help="Markdown alert output path.", ) subparser.add_argument( "--jsonl-path", default=str(DEFAULT_JSONL), help="JSONL alert output path.", ) subparser.add_argument( "--event-watchlist-path", default=str(DEFAULT_EVENT_WATCHLIST), help="Output path for the dynamic event-driven watchlists JSON.", ) subparser.add_argument( "--report-hours", type=int, default=24, help="Lookback window for the markdown snapshot report.", ) subparser.add_argument("--format", choices=["markdown", "json"], default="markdown") poll = subparsers.add_parser("poll", help="Fetch feeds once and store new alerts.") add_common_io(poll) poll.set_defaults(func=run_poll) loop = subparsers.add_parser("loop", help="Continuously fetch feeds on an interval.") add_common_io(loop) loop.add_argument("--interval-seconds", type=int, default=300, help="Polling interval in seconds.") loop.set_defaults(func=run_loop) report = subparsers.add_parser("report", help="Render a report from stored alerts.") report.add_argument( "--db-path", default=str(DEFAULT_DB), help="SQLite database path. Defaults under the state directory.", ) report.add_argument( "--event-watchlist-path", default=str(DEFAULT_EVENT_WATCHLIST), help="Optional output path for the dynamic event-driven watchlists JSON.", ) report.add_argument("--hours", type=int, default=12, help="Lookback window in hours.") report.set_defaults(func=run_report) return parser def main() -> int: parser = build_parser() args = parser.parse_args() state_dir = Path(args.state_dir) ensure_state_dir(state_dir) if getattr(args, "db_path", None) == str(DEFAULT_DB): args.db_path = str(state_dir / DEFAULT_DB.name) if getattr(args, "markdown_path", None) == str(DEFAULT_MARKDOWN): args.markdown_path = str(state_dir / DEFAULT_MARKDOWN.name) if getattr(args, "jsonl_path", None) == str(DEFAULT_JSONL): args.jsonl_path = str(state_dir / DEFAULT_JSONL.name) if getattr(args, "event_watchlist_path", None) == str(DEFAULT_EVENT_WATCHLIST): args.event_watchlist_path = str(state_dir / DEFAULT_EVENT_WATCHLIST.name) return args.func(args) if __name__ == "__main__": sys.exit(main()) FILE:scripts/opening_window_checklist.py #!/usr/bin/env python3 from __future__ import annotations import argparse import json from pathlib import Path from capital_flow import ( attach_flow_tags, build_flow_lookup, build_group_flow_scoreboard, fetch_market_flow_snapshot, fetch_top_main_flows, render_flow_snapshot, ) from industry_chain import enrich_event_payload_with_chain_focus from market_data import fetch_tencent_quotes, format_markdown_table from market_sentiment import build_sentiment_snapshot ROOT = Path(__file__).resolve().parents[1] DEFAULT_WATCHLIST = ROOT / "assets" / "default_watchlists.json" DEFAULT_EVENT_WATCHLIST = Path.home() / ".uwillberich" / "news-iterator" / "event_watchlists.json" EVENT_CATEGORY_ORDER = ["huge_conflict", "huge_future", "huge_name_release"] CATEGORY_LABELS = { "huge_conflict": "巨大冲突", "huge_future": "巨大前景", "huge_name_release": "巨头名人", } SIGNAL_LABELS = {"high": "高", "medium": "中", "low": "低"} KEYWORD_LABELS = { "war": "战争", "oil": "原油", "energy": "能源", "chips": "芯片", "chip": "芯片", "robots": "机器人", "robot": "机器人", "launch": "发布", "launches": "发布", "announces": "宣布", "announce": "宣布", "unveils": "亮相", "unveil": "亮相", "data center": "数据中心", } TIME_GATES = [ { "time": "09:00", "watch": "LPR and policy timing", "bullish": "5Y LPR cut or clearly supportive policy tone", "bearish": "No support and policy-sensitive names stay weak", }, { "time": "09:20-09:25", "watch": "Auction leadership", "bullish": "Tech repair groups lead the bid", "bearish": "Only oil, coal, banks, or telecom lead", }, { "time": "09:30-10:00", "watch": "Prior-close reclaim and index support", "bullish": "Core leaders reclaim prior close and broad indices stabilize", "bearish": "Leaders stay under prior close and defensives dominate", }, { "time": "10:00-10:30", "watch": "Breadth expansion", "bullish": "Repair broadens beyond 2-3 names", "bearish": "Bounce stays narrow and fades", }, ] def category_display_name(category: str) -> str: return CATEGORY_LABELS.get(category, category) def signal_display_name(signal: str) -> str: return SIGNAL_LABELS.get(signal, signal) def format_keyword_list(keywords: list[str]) -> str: if not keywords: return "n/a" return ", ".join(KEYWORD_LABELS.get(keyword, keyword) for keyword in keywords) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Build a first-30-minute A-share opening checklist.") parser.add_argument( "--watchlist", default=str(DEFAULT_WATCHLIST), help="Path to watchlist JSON. Defaults to the bundled watchlist.", ) parser.add_argument( "--groups", nargs="+", default=["tech_repair", "policy_beta", "defensive_gauge"], help="Watchlist groups to score during the opening window.", ) parser.add_argument( "--event-watchlist", default=str(DEFAULT_EVENT_WATCHLIST), help="Path to dynamic event-driven watchlists JSON.", ) parser.add_argument( "--skip-event-pools", action="store_true", help="Do not append event-driven watchlists from the news iterator state.", ) parser.add_argument( "--skip-capital-flow", action="store_true", help="Do not append capital-flow overlays.", ) parser.add_argument( "--skip-sentiment", action="store_true", help="Do not append the market-sentiment overlay.", ) parser.add_argument( "--skip-industry-chain", action="store_true", help="Do not enrich event pools with chain-focus groups.", ) return parser def load_watchlist(path: str) -> dict: return json.loads(Path(path).read_text(encoding="utf-8")) def load_event_payload(path: str) -> dict: event_path = Path(path) if not event_path.exists(): return {} return json.loads(event_path.read_text(encoding="utf-8")) def build_signal_lookup(watchlist: dict) -> dict[str, dict]: lookup: dict[str, dict] = {} for items in watchlist.values(): for item in items: symbol = item["symbol"] strong_signal = item.get("strong_signal") weak_signal = item.get("weak_signal") if not strong_signal and not weak_signal: continue lookup[symbol] = { "strong_signal": strong_signal or "", "weak_signal": weak_signal or "", } return lookup def summarize_group(items: list[dict], quotes: list[dict]) -> dict: quote_map = {quote["code"]: quote for quote in quotes} above = 0 below = 0 flat = 0 changes = [] for item in items: code = item["symbol"][2:] quote = quote_map.get(code) if not quote or quote.get("change_pct") is None: continue changes.append(quote["change_pct"]) if (quote.get("price") or 0) > (quote.get("prev_close") or 0): above += 1 elif (quote.get("price") or 0) < (quote.get("prev_close") or 0): below += 1 else: flat += 1 avg_change = round(sum(changes) / len(changes), 2) if changes else None return { "group": "", "count": len(items), "above_prev_close": above, "below_prev_close": below, "flat": flat, "avg_change_pct": avg_change, } def classify_state(scoreboard: list[dict]) -> str: by_name = {row["group"]: row for row in scoreboard} tech = by_name.get("tech_repair", {}) policy = by_name.get("policy_beta", {}) defensive = by_name.get("defensive_gauge", {}) tech_above = tech.get("above_prev_close", 0) defensive_above = defensive.get("above_prev_close", 0) policy_above = policy.get("above_prev_close", 0) if tech_above >= 3 and defensive_above <= 2: return "State: likely true repair" if policy_above >= 2 and tech_above >= 2: return "State: likely policy-backed repair" if defensive_above >= 3 and tech_above <= 2: return "State: likely defensive concentration" return "State: mixed or unresolved opening tape" def build_detail_rows(items: list[dict], quotes: list[dict], signal_lookup: dict[str, dict], flow_lookup: dict[str, dict]) -> list[dict]: quote_map = {quote["code"]: quote for quote in quotes} rows = [] for item in items: code = item["symbol"][2:] quote = quote_map.get(code) if not quote: continue fallback = signal_lookup.get(item["symbol"], {}) rows.append( { "name": quote["name"], "code": quote["code"], "role": item["role"], "price": quote["price"], "chg%": quote["change_pct"], "event_score": item.get("event_score"), "trigger_count": item.get("trigger_count"), "event_driver": item.get("event_driver", ""), "strong_signal": item.get("strong_signal") or fallback.get("strong_signal", ""), "weak_signal": item.get("weak_signal") or fallback.get("weak_signal", ""), } ) return attach_flow_tags(rows, flow_lookup) def render_detail_table(rows: list[dict], is_event: bool) -> str: columns = [ ("Name", "name"), ("Code", "code"), ("Role", "role"), ("Price", "price"), ("Chg%", "chg%"), ("FlowTag", "flow_tag"), ("Flow(亿)", "flow_yi"), ] if is_event: columns.extend( [ ("EventScore", "event_score"), ("Triggers", "trigger_count"), ("Driver", "event_driver"), ] ) columns.extend( [ ("Strong Signal", "strong_signal"), ("Weak Signal", "weak_signal"), ] ) return format_markdown_table(rows, columns) def render_event_summary(payload: dict) -> None: summary = payload.get("summary", []) if not summary: return rows = [ { "category": category_display_name(item["category"]), "alert_count": item["alert_count"], "total_score": item["total_score"], "top_keywords": format_keyword_list(item.get("top_keywords", [])), } for item in summary ] print("\n## 事件驱动层总结") print( format_markdown_table( rows, [ ("类别", "category"), ("条数", "alert_count"), ("总分", "total_score"), ("高频关键词", "top_keywords"), ], ) ) def render_event_top_alerts(payload: dict) -> None: top_alerts = payload.get("top_alerts", {}) if not top_alerts: return print("\n## 事件信息源链接") for category in EVENT_CATEGORY_ORDER: items = top_alerts.get(category, []) if not items: continue print(f"\n### {category_display_name(category)} Top 10 信息源") for index, item in enumerate(items, start=1): print(f"{index}. [{item['title']}]({item['link']})") print( f" - 来源: {item['source']} | 信号: `{signal_display_name(item['signal'])}` | 分值: `{item['score']}`" ) print(f" - 实体: {', '.join(item.get('entities', [])) or 'n/a'}") print(f" - 关键词: {format_keyword_list(item.get('keywords', []))}") def render_chain_summary(payload: dict) -> None: summary = payload.get("chain_summary", []) if not summary: return rows = [ { "theme": item["theme"], "score": item["score"], "group": item["group"], "reasons": " / ".join(item.get("reasons", [])[:3]) or "n/a", } for item in summary ] print("\n## Industry Chain Focus") print( format_markdown_table( rows, [ ("Theme", "theme"), ("Score", "score"), ("Group", "group"), ("Reasons", "reasons"), ], ) ) def classify_event_overlay(scoreboard: list[dict]) -> str: by_name = {row["group"]: row for row in scoreboard} conflict_benefit = by_name.get("event_focus_huge_conflict_benefit", by_name.get("event_focus_huge_conflict", {})) conflict_headwind = by_name.get("event_focus_huge_conflict_headwind", {}) future = by_name.get("event_focus_huge_future", {}) name_release = by_name.get("event_focus_huge_name_release", {}) conflict_benefit_above = conflict_benefit.get("above_prev_close", 0) conflict_headwind_below = conflict_headwind.get("below_prev_close", 0) future_above = future.get("above_prev_close", 0) release_above = name_release.get("above_prev_close", 0) if conflict_benefit_above >= 4 and conflict_headwind_below >= 3: return "Event Overlay: conflict beneficiaries are confirming while compute-power names stay under pressure." if future_above >= 4 or release_above >= 4: return "Event Overlay: future/big-name news is translating into tradeable technology leadership." return "Event Overlay: messages are present, but translation into price action is still mixed." def classify_opening_bias(scoreboard: list[dict], group_flow_rows: list[dict], sentiment: dict | None) -> str: base_read = classify_state(scoreboard) if not sentiment: return base_read by_name = {row["group"]: row for row in group_flow_rows} tech_flow = float((by_name.get("tech_repair") or {}).get("net_flow_yi") or 0) defensive_flow = float((by_name.get("defensive_gauge") or {}).get("net_flow_yi") or 0) if "defensive concentration" in base_read.lower() and sentiment.get("label") == "抱团行情": return "Open Read: 抱团行情延续，优先把油气、煤炭、红利当环境锚。" if "true repair" in base_read.lower() and tech_flow > 0: return "Open Read: 价格与资金共振，科技修复可信度提升。" if tech_flow > defensive_flow and sentiment.get("label") in {"科技修复", "修复扩散"}: return "Open Read: 资金更偏向成长，优先跟修复扩散而不是防御抱团。" return f"Open Read: {sentiment.get('read', base_read)}" def main() -> None: parser = build_parser() args = parser.parse_args() watchlist = load_watchlist(args.watchlist) event_payload = {} if args.skip_event_pools else load_event_payload(args.event_watchlist) if event_payload and not args.skip_industry_chain: event_payload = enrich_event_payload_with_chain_focus( event_payload, watchlist, selected_groups=args.groups, ) event_groups = event_payload.get("groups", {}) signal_lookup = build_signal_lookup(watchlist) selected_groups = [group for group in args.groups if group in watchlist] selected_event_groups = [group for group in args.groups if group in event_groups] if not selected_event_groups and event_groups: selected_event_groups = event_payload.get("default_report_groups", []) selected_event_groups = list(dict.fromkeys(selected_event_groups)) all_symbols = [] for group in selected_groups: all_symbols.extend(item["symbol"] for item in watchlist[group]) for group in selected_event_groups: all_symbols.extend(item["symbol"] for item in event_groups.get(group, [])) quotes = fetch_tencent_quotes(dict.fromkeys(all_symbols)) flow_lookup: dict[str, dict] = {} group_flow_rows: list[dict] = [] market_flow_rows: list[dict] = [] if not args.skip_capital_flow: market_flow = fetch_market_flow_snapshot() inflow_items = fetch_top_main_flows("inflow", limit=8) outflow_items = fetch_top_main_flows("outflow", limit=8) flow_lookup = build_flow_lookup(inflow_items, outflow_items) group_flow_rows = build_group_flow_scoreboard(watchlist, selected_groups, flow_lookup) market_flow_rows = render_flow_snapshot(market_flow) sentiment = None if args.skip_sentiment else build_sentiment_snapshot(group_flow_rows=group_flow_rows) print("# Opening Window Checklist") print() print("## Time Gates") print( format_markdown_table( TIME_GATES, [ ("Time", "time"), ("Watch", "watch"), ("Bullish Read", "bullish"), ("Bearish Read", "bearish"), ], ) ) scoreboard = [] for group in selected_groups: summary = summarize_group(watchlist[group], quotes) summary["group"] = group scoreboard.append(summary) print("\n## Group Scoreboard") print( format_markdown_table( scoreboard, [ ("Group", "group"), ("Count", "count"), ("Above Prev Close", "above_prev_close"), ("Below Prev Close", "below_prev_close"), ("Flat", "flat"), ("Avg Chg%", "avg_change_pct"), ], ) ) print("\n## Quick Read") print(classify_state(scoreboard)) if market_flow_rows: print("\n## Capital Flow Snapshot") print( format_markdown_table( market_flow_rows, [ ("State", "label"), ("MainNet(亿)", "main_net_yi"), ("BigInflow(亿)", "big_order_inflow_yi"), ("MediumInflow(亿)", "medium_order_inflow_yi"), ("SmallInflow(亿)", "small_order_inflow_yi"), ("As Of", "as_of"), ], ) ) if group_flow_rows: print("\n## Capital Flow Scoreboard") print( format_markdown_table( group_flow_rows, [ ("Group", "group"), ("InflowHits", "inflow_hits"), ("OutflowHits", "outflow_hits"), ("NetFlow(亿)", "net_flow_yi"), ("Bias", "bias"), ("Leaders", "leaders"), ], ) ) if sentiment: print("\n## Sentiment Read") print(f"- state: {sentiment['label']}") print(f"- read: {sentiment['read']}") print(f"- opening_bias: {classify_opening_bias(scoreboard, group_flow_rows, sentiment)}") for group in selected_groups: rows = build_detail_rows(watchlist[group], quotes, signal_lookup, flow_lookup) print(f"\n## Watchlist: {group}") print(render_detail_table(rows, is_event=False)) if event_groups and selected_event_groups: render_event_summary(event_payload) render_event_top_alerts(event_payload) render_chain_summary(event_payload) event_scoreboard = [] for group in selected_event_groups: summary = summarize_group(event_groups[group], quotes) summary["group"] = group event_scoreboard.append(summary) print("\n## Event Overlay Scoreboard") print( format_markdown_table( event_scoreboard, [ ("Group", "group"), ("Count", "count"), ("Above Prev Close", "above_prev_close"), ("Below Prev Close", "below_prev_close"), ("Flat", "flat"), ("Avg Chg%", "avg_change_pct"), ], ) ) print("\n## Event Overlay Read") print(classify_event_overlay(event_scoreboard)) for group in selected_event_groups: rows = build_detail_rows(event_groups[group], quotes, signal_lookup, flow_lookup) print(f"\n## Event Watchlist: {group}") print(render_detail_table(rows, is_event=True)) if __name__ == "__main__": main() FILE:scripts/runtime_config.py #!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import stat import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] DEFAULT_RUNTIME_HOME = Path.home() / ".uwillberich" LEGACY_RUNTIME_HOME = Path.home() / ".a-share-decision-desk" DEFAULT_ENV_PATH = DEFAULT_RUNTIME_HOME / "runtime.env" LEGACY_ENV_PATH = LEGACY_RUNTIME_HOME / "runtime.env" DEFAULT_EXAMPLE_ENV = ROOT / "assets" / "runtime.env.example" DEFAULT_DATA_DIR = DEFAULT_RUNTIME_HOME / "data" RUNTIME_ENV_VARS = ("UWILLBERICH_RUNTIME_ENV", "A_SHARE_RUNTIME_ENV") DATA_DIR_ENV_VARS = ("UWILLBERICH_DATA_DIR", "A_SHARE_DECISION_DATA_DIR") OPTIONAL_KEYS = ("EM_API_KEY",) EM_INTEGRATIONS = ("MX_FinSearch", "MX_StockPick", "MX_MacroData", "MX_FinData") EASTMONEY_APPLY_URL = "https://ai.eastmoney.com/mxClaw" EASTMONEY_HOME_URL = "https://ai.eastmoney.com/nlink/" def parse_env_text(text: str) -> dict[str, str]: values: dict[str, str] = {} for raw_line in text.splitlines(): line = raw_line.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line[7:].strip() if "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip() if not key: continue if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}: value = value[1:-1] values[key] = value return values def read_env_file(path: Path) -> dict[str, str]: if not path.exists(): return {} return parse_env_text(path.read_text(encoding="utf-8")) def resolve_env_paths(env_path: str | None = None) -> list[Path]: paths: list[Path] = [] if env_path: paths.append(Path(env_path).expanduser()) for env_var in RUNTIME_ENV_VARS: custom = os.environ.get(env_var) if custom: paths.append(Path(custom).expanduser()) paths.extend([DEFAULT_ENV_PATH, LEGACY_ENV_PATH, ROOT / ".env.local", ROOT / ".env"]) deduped: list[Path] = [] seen: set[str] = set() for path in paths: resolved = str(path.expanduser()) if resolved in seen: continue seen.add(resolved) deduped.append(Path(resolved)) return deduped def load_runtime_env(env_path: str | None = None, override: bool = False) -> dict[str, str]: loaded: dict[str, str] = {} for path in resolve_env_paths(env_path): values = read_env_file(path) for key, value in values.items(): if override or key not in os.environ: os.environ[key] = value loaded[key] = value em_key = os.environ.get("EM_API_KEY", "").strip() mx_key = os.environ.get("MX_APIKEY", "").strip() if em_key and not mx_key: os.environ["MX_APIKEY"] = em_key loaded.setdefault("MX_APIKEY", em_key) elif mx_key and not em_key: os.environ["EM_API_KEY"] = mx_key loaded.setdefault("EM_API_KEY", mx_key) return loaded def redact_value(value: str) -> str: if not value: return "" if len(value) <= 8: return "*" * len(value) return f"{value[:4]}...{value[-4:]}" def build_capabilities() -> dict[str, object]: em_ready = bool(os.environ.get("EM_API_KEY") or os.environ.get("MX_APIKEY")) return { "public_mode": False, "em_required_mode": True, "em_key_configured": em_ready, "em_enhanced_mode": em_ready, "available_integrations": list(EM_INTEGRATIONS) if em_ready else [], } def em_key_setup_instructions(script_hint: str | None = None) -> str: hint = script_hint or "python3 scripts/runtime_config.py set-em-key --stdin" return ( "EM_API_KEY is required for uwillberich.\n" f"Apply here: {EASTMONEY_APPLY_URL}\n" "After opening the link, click download and you will see the key.\n" f"Official site: {EASTMONEY_HOME_URL}\n" "Store the key in ~/.uwillberich/runtime.env, or run:\n" f"printf '%s' 'your_em_api_key' | {hint}" ) def require_em_api_key(env_path: str | None = None, script_hint: str | None = None) -> str: load_runtime_env(env_path) key = (os.environ.get("EM_API_KEY") or os.environ.get("MX_APIKEY") or "").strip() if key: return key raise RuntimeError(em_key_setup_instructions(script_hint)) def get_output_root() -> Path: load_runtime_env() custom = "" for env_var in DATA_DIR_ENV_VARS: value = (os.environ.get(env_var) or "").strip() if value: custom = value break path = Path(custom).expanduser() if custom else DEFAULT_DATA_DIR path.mkdir(parents=True, exist_ok=True) return path def get_output_dir(subdir: str | None = None) -> Path: root = get_output_root() if not subdir: return root path = root / subdir path.mkdir(parents=True, exist_ok=True) return path def build_status(env_path: str | None = None) -> dict[str, object]: load_runtime_env(env_path) env_paths = resolve_env_paths(env_path) existing_path = next((str(path) for path in env_paths if path.exists()), "") configured_keys = [key for key in OPTIONAL_KEYS if os.environ.get(key)] if os.environ.get("MX_APIKEY") and "EM_API_KEY" not in configured_keys: configured_keys.append("EM_API_KEY") return { "runtime_env_path": existing_path or str(env_paths[0]), "env_file_exists": bool(existing_path), "configured_keys": configured_keys, "redacted_values": {key: redact_value(os.environ.get(key, "")) for key in configured_keys}, "capabilities": build_capabilities(), "example_env_path": str(DEFAULT_EXAMPLE_ENV), "output_root": str(get_output_root()), "eastmoney_apply_url": EASTMONEY_APPLY_URL, } def write_env_file(path: Path, values: dict[str, str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) lines = [f"{key}={value}" for key, value in sorted(values.items())] path.write_text("\n".join(lines) + "\n", encoding="utf-8") path.chmod(stat.S_IRUSR | stat.S_IWUSR) def set_em_key(path: Path, value: str) -> None: values = read_env_file(path) values["EM_API_KEY"] = value.strip() write_env_file(path, values) os.environ["EM_API_KEY"] = value.strip() os.environ["MX_APIKEY"] = value.strip() def unset_em_key(path: Path) -> None: values = read_env_file(path) values.pop("EM_API_KEY", None) write_env_file(path, values) os.environ.pop("EM_API_KEY", None) os.environ.pop("MX_APIKEY", None) def print_status(env_path: str | None, as_json: bool) -> int: status = build_status(env_path) if as_json: print(json.dumps(status, ensure_ascii=False, indent=2)) return 0 print(f"runtime_env_path: {status['runtime_env_path']}") print(f"env_file_exists: {status['env_file_exists']}") print(f"configured_keys: {', '.join(status['configured_keys']) or 'none'}") print(f"em_required_mode: {status['capabilities']['em_required_mode']}") print(f"em_key_configured: {status['capabilities']['em_key_configured']}") integrations = status["capabilities"]["available_integrations"] print(f"available_integrations: {', '.join(integrations) or 'none until EM_API_KEY is configured'}") print(f"eastmoney_apply_url: {status['eastmoney_apply_url']}") print(f"output_root: {status['output_root']}") return 0 def run_set_em_key(args: argparse.Namespace) -> int: value = args.value if args.stdin: value = sys.stdin.read().strip() if not value: print("missing --value or --stdin", file=sys.stderr) return 1 env_path = Path(args.env_path).expanduser() set_em_key(env_path, value) print(f"stored_em_api_key: {env_path}") print(f"em_enhanced_mode: {build_capabilities()['em_enhanced_mode']}") return 0 def run_unset_em_key(args: argparse.Namespace) -> int: env_path = Path(args.env_path).expanduser() unset_em_key(env_path) print(f"removed_em_api_key: {env_path}") return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Manage local runtime credentials for uwillberich.") parser.add_argument( "--env-path", default=str(DEFAULT_ENV_PATH), help="Runtime env file path. Defaults to ~/.uwillberich/runtime.env", ) subparsers = parser.add_subparsers(dest="command", required=True) status_parser = subparsers.add_parser("status", help="Show credential and capability status.") status_parser.add_argument("--json", action="store_true", help="Render status as JSON.") status_parser.set_defaults(func=lambda args: print_status(args.env_path, args.json)) set_parser = subparsers.add_parser("set-em-key", help="Store EM_API_KEY in the local runtime env file.") set_parser.add_argument("--value", help="EM_API_KEY value.") set_parser.add_argument("--stdin", action="store_true", help="Read EM_API_KEY from stdin.") set_parser.set_defaults(func=run_set_em_key) unset_parser = subparsers.add_parser("unset-em-key", help="Remove EM_API_KEY from the local runtime env file.") unset_parser.set_defaults(func=run_unset_em_key) return parser def main() -> int: parser = build_parser() args = parser.parse_args() return args.func(args) if __name__ == "__main__": sys.exit(main()) FILE:scripts/smoke_test.py #!/usr/bin/env python3 from __future__ import annotations import json import sys import tempfile from pathlib import Path from market_data import fetch_index_snapshot, fetch_sector_movers, fetch_tencent_quotes from industry_chain import load_json as load_chain_json, select_chain_themes from market_sentiment import build_sentiment_snapshot from mx_toolkit import load_presets from news_iterator import FeedItem, build_event_watchlists_payload, classify_item from opening_window_checklist import classify_state from runtime_config import build_status, get_output_dir, read_env_file ROOT = Path(__file__).resolve().parents[1] def assert_true(condition: bool, message: str) -> None: if not condition: raise AssertionError(message) def normalize_alert(alert: dict) -> dict: return { "category": alert["category"], "score": alert["score"], "watchlists": alert.get("impacted_watchlists", []), "watchlist_scores": alert.get("watchlist_scores", {}), "entities": alert.get("matched_entities", []), "keywords": alert.get("matched_keywords", []), } def main() -> None: indices = fetch_index_snapshot() assert_true(len(indices) >= 3, "expected at least 3 indices") assert_true(any(item.get("name") == "上证指数" for item in indices), "missing 上证指数") leaders = fetch_sector_movers(limit=3, rising=True) laggards = fetch_sector_movers(limit=3, rising=False) assert_true(len(leaders) == 3, "expected 3 top sectors") assert_true(len(laggards) == 3, "expected 3 bottom sectors") quotes = fetch_tencent_quotes(["sz300502", "sh688981", "sh600938"]) assert_true(len(quotes) == 3, "expected 3 quotes") assert_true(all(quote.get("price") is not None for quote in quotes), "quote price missing") watchlists = json.loads((ROOT / "assets" / "default_watchlists.json").read_text(encoding="utf-8")) iterator_config = json.loads((ROOT / "assets" / "news_iterator_config.json").read_text(encoding="utf-8")) chain_config = load_chain_json(str(ROOT / "assets" / "industry_chains.json")) mx_presets = load_presets(str(ROOT / "assets" / "mx_presets.json")) assert_true("cross_cycle_anchor12" in watchlists, "missing cross_cycle_anchor12") assert_true("cross_cycle_core" in watchlists, "missing cross_cycle_core") assert_true("war_shock_core12" in watchlists, "missing war_shock_core12") assert_true("war_benefit_oil_coal" in watchlists, "missing war_benefit_oil_coal") assert_true("war_headwind_compute_power" in watchlists, "missing war_headwind_compute_power") assert_true(len(watchlists["cross_cycle_anchor12"]) >= 10, "anchor watchlist too small") assert_true("feeds" in iterator_config and len(iterator_config["feeds"]) >= 5, "news iterator feeds missing") assert_true("conflict_entities" in iterator_config, "news iterator conflict entities missing") assert_true(len(chain_config.get("themes", [])) >= 5, "industry chain config missing themes") assert_true("preopen_policy" in mx_presets, "missing MX preset preopen_policy") assert_true("preopen_repair_chain" in mx_presets, "missing MX preset preopen_repair_chain") with tempfile.TemporaryDirectory() as temp_dir: env_path = Path(temp_dir) / "runtime.env" env_path.write_text("EM_API_KEY=test-key\n", encoding="utf-8") env_values = read_env_file(env_path) assert_true(env_values.get("EM_API_KEY") == "test-key", "runtime env parsing failed") status = build_status(str(env_path)) assert_true(status["capabilities"]["em_required_mode"], "EM key should be mandatory") assert_true(status["capabilities"]["em_enhanced_mode"], "runtime capability detection failed") assert_true("EM_API_KEY" in status["configured_keys"], "runtime key status missing") assert_true(status["eastmoney_apply_url"].startswith("https://ai.eastmoney.com/"), "missing Eastmoney apply url") output_dir = get_output_dir("smoke-test-output") assert_true(output_dir.exists(), "runtime output dir missing") future_release_alerts = classify_item( FeedItem( item_key="future", feed_key="test", feed_label="Test", source="Test", title="OpenAI unveiled a new model for datacenter reasoning agents", link="https://example.com/future", summary="The launch centers on AI server demand and semiconductor inference.", published_at="2026-03-19T00:00:00+00:00", ), iterator_config, ) categories = {alert["category"] for alert in future_release_alerts} assert_true("huge_future" in categories, "expected huge_future classification") assert_true("huge_name_release" in categories, "expected huge_name_release classification") conflict_alerts = classify_item( FeedItem( item_key="conflict", feed_key="test", feed_label="Test", source="Test", title="Iran attack raises oil shipping disruption risk in Hormuz", link="https://example.com/conflict", summary="Energy traders watch crude, refinery routes and power costs for data center operators.", published_at="2026-03-19T00:00:00+00:00", ), iterator_config, ) conflict_categories = {alert["category"] for alert in conflict_alerts} assert_true("huge_conflict" in conflict_categories, "expected huge_conflict classification") assert_true( any("war_benefit_oil_coal" in alert["impacted_watchlists"] for alert in conflict_alerts), "expected war_benefit_oil_coal mapping", ) payload = build_event_watchlists_payload( [normalize_alert(alert) for alert in future_release_alerts + conflict_alerts], watchlists, hours=12, ) assert_true("event_focus_core" in payload["groups"], "missing event_focus_core") assert_true("event_focus_huge_conflict_benefit" in payload["groups"], "missing conflict benefit pool") assert_true( any(item["symbol"] == "sh600938" for item in payload["groups"]["event_focus_huge_conflict_benefit"]), "expected China Offshore Oil in conflict event pool", ) chain_themes = select_chain_themes(payload, ["tech_repair", "defensive_gauge"], chain_config, max_themes=3) chain_ids = {item["id"] for item in chain_themes} assert_true("optical_module_chain" in chain_ids, "expected optical-module chain theme") assert_true("oil_gas_chain" in chain_ids or "coal_chain" in chain_ids, "expected energy chain theme") state = classify_state( [ {"group": "tech_repair", "above_prev_close": 3}, {"group": "policy_beta", "above_prev_close": 1}, {"group": "defensive_gauge", "above_prev_close": 1}, ] ) assert_true("true repair" in state.lower(), "opening-window classifier mismatch") sentiment = build_sentiment_snapshot( group_flow_rows=[ {"group": "tech_repair", "net_flow_yi": 6.5}, {"group": "defensive_gauge", "net_flow_yi": -2.3}, ] ) assert_true(sentiment["label"] in {"科技修复", "修复扩散", "分化震荡", "抱团行情", "分化偏弱"}, "unexpected sentiment label") print("smoke test passed") print(f"indices: {len(indices)}") print(f"leaders: {len(leaders)}") print(f"laggards: {len(laggards)}") print(f"quotes: {len(quotes)}") if __name__ == "__main__": try: main() except Exception as exc: print(f"smoke test failed: {exc}", file=sys.stderr) raise

uwillberich

Comments (0)