源代码.md 缠论股票分析软件 —— 源代码 本文件由 generate_src_md.py 自动汇总项目自研源代码生成。 第三方依赖（vendor/chanpy 即 chan.py 引擎）未逐文件展开，仅作依赖说明。

目录 app.py run_app.py chan_stock.spec pyproject.toml pages/single_stock.py pages/watchlist.py tests/test_config.py src/init.py src/chan_stock/init.py src/chan_stock/config.py src/chan_stock/engine/init.py src/chan_stock/engine/analyzer.py src/chan_stock/logging_setup.py src/chan_stock/main.py src/chan_stock/settings.py app.py “”“Streamlit 入口：导航到 个股分析 / 自选股 页面。”“” from future import annotations

import sys from pathlib import Path

ROOT = Path(file).resolve().parent sys.path.insert(0, str(ROOT / “src”))

import streamlit as st

st.set_page_config(page_title=“缠论股票分析”, layout=“wide”)

汉化 Streamlit 框架自带 UI（右上角菜单、底部水印等）

_UI_I18N = “”” <style> /* 隐藏底部 “Made with Streamlit” 水印 */ footer { visibility: hidden; height: 0; } </style> <script> (function () { const i18n = { ‘Deploy’: ‘部署’, ‘System’: ‘跟随系统’, ‘Light’: ‘浅色’, ‘Dark’: ‘深色’, ‘Rerun’: ‘重新运行’, ‘Auto rerun’: ‘自动重新运行’, ‘Clear cache’: ‘清除缓存’, ‘Print’: ‘打印’, ‘Record screen’: ‘录屏’, ‘Made with Streamlit’: ‘由 Streamlit 驱动’ }; function walk(node) { if (node.nodeType === Node.TEXT_NODE) { for (const [en, zh] of Object.entries(i18n)) { if (node.textContent.includes(en)) { node.textContent = node.textContent.split(en).join(zh); } } } else { Array.from(node.childNodes).forEach(walk); } } walk(document.body); const obs = new MutationObserver((muts) => { for (const m of muts) { if (m.type === ‘childList’) { Array.from(m.addedNodes).forEach((n) => { if (n.nodeType === 1) walk(n); }); } else if (m.type === ‘characterData’) { walk(m.target); } } }); obs.observe(document.body, { childList: true, subtree: true, characterData: true }); })(); </script> “”” st.markdown(_UI_I18N, unsafe_allow_html=True)

single = st.Page(“pages/single_stock.py”, title=“个股分析”, default=True) watch = st.Page(“pages/watchlist.py”, title=“自选股”)

pg = st.navigation([single, watch]) pg.run() run_app.py “”“PyInstaller 入口：启动 Streamlit 服务并自动打开浏览器。

等价于 streamlit run app.py，但适配冻结环境：

• 资源位于 sys._MEIPASS，运行时数据落到用户 AppData。
• 严格复刻 streamlit 的 _main_run 初始化（设置 _config._main_script_path、 load_config_options），否则 / 路由不会注册，浏览器打开会得到 404。 “”” from future import annotations

import os import sys import threading import time import webbrowser from pathlib import Path

import streamlit from streamlit import config as _config import streamlit.web.bootstrap as bootstrap from streamlit.web.server.app_discovery import discover_asgi_app

def _open_browser() -> None: # 等服务起来再打开浏览器 time.sleep(2.5) try: webbrowser.open(“http://localhost:8501”) except Exception: # noqa: BLE001 pass

def _base() -> Path: if getattr(sys, “frozen”, False): return Path(sys._MEIPASS) return Path(file).resolve().parent

def main() -> None: os.environ.setdefault(“STREAMLIT_SERVER_HEADLESS”, “true”) os.environ[“STREAMLIT_BROWSER_GATHER_USAGE_STATS”] = “false” os.environ.setdefault(“STREAMLIT_SERVER_PORT”, “8501”)

base = _base()
app_path = str(base / "app.py")
main_script_path = os.path.abspath(app_path)

flag_options = {"server.headless": True}
# 复刻 streamlit _main_run 的关键初始化（否则 / 路由不注册 -> 404）
_config._main_script_path = main_script_path
bootstrap.load_config_options(flag_options=flag_options)

# Python 冻结后 streamlit.__file__ 路径不再含 "site-packages"，
# 导致 global.developmentMode 被推断为 True，进而跳过 / 静态资源路由（index.html），
# 浏览器打开会得到 404。强制回生产模式（False）以恢复根路由注册。
_config.set_option("global.developmentMode", False)

threading.Thread(target=_open_browser, daemon=True).start()

discovery = discover_asgi_app(Path(main_script_path))
if discovery.is_asgi_app:
    bootstrap.run_asgi_app(main_script_path, discovery.import_string, [], flag_options)
else:
    bootstrap.run(main_script_path, False, [], flag_options)

if name == “main”: main() chan_stock.spec

-- mode: python ; coding: utf-8 --

from PyInstaller.utils.hooks import collect_all

datas = [(‘app.py’, ‘.’), (‘pages’, ‘pages’), (‘build_stage/vendor/chanpy’, ‘vendor/chanpy’), (‘src/chan_stock’, ‘chan_stock’)] binaries = [] hiddenimports = [‘pytdx’, ‘pytdx.hq’, ‘DataAPI.myapi’] tmp_ret = collect_all(‘streamlit’) datas += tmp_ret[0]; binaries += tmp_ret[1]; hiddenimports += tmp_ret[2] tmp_ret = collect_all(‘matplotlib’) datas += tmp_ret[0]; binaries += tmp_ret[1]; hiddenimports += tmp_ret[2]

a = Analysis( [‘run_app.py’], pathex=[‘vendor/chanpy’, ‘src’], binaries=binaries, datas=datas, hiddenimports=hiddenimports, hookspath=[], hooksconfig={}, runtime_hooks=[], excludes=[‘akshare’, ‘curl_cffi’], noarchive=False, optimize=0, ) pyz = PYZ(a.pure)

exe = EXE( pyz, a.scripts, [], exclude_binaries=True, name=‘chan_stock’, debug=False, bootloader_ignore_signals=False, strip=False, upx=True, console=True, disable_windowed_traceback=False, argv_emulation=False, target_arch=None, codesign_identity=None, entitlements_file=None, ) coll = COLLECT( exe, a.binaries, a.datas, strip=False, upx=True, upx_exclude=[], name=‘chan_stock’, ) pyproject.toml [project] name = “chan_stock” version = “0.1.0” description = “缠论股票分析软件” readme = “README.md” requires-python = “>=3.12” dependencies = [ “python-dotenv>=1.0.1”, “streamlit>=1.30”, “pandas>=2.0”, “numpy>=1.24”, “matplotlib>=3.5”, “akshare>=1.18”, “pytdx>=1.72”, ]

[project.scripts] chanstock = “chan_stock.main:main”

[build-system] requires = [“hatchling”] build-backend = “hatchling.build”

[tool.ruff] line-length = 100 target-version = “py312”

[tool.ruff.lint] select = [“E”, “F”, “I”, “N”, “UP”]

[tool.pytest.ini_options] testpaths = [“tests”]

[tool.mypy] python_version = “3.12” ignore_missing_imports = true warn_return_any = true warn_unused_configs = true pages/single_stock.py “”“个股缠论分析页。”“” from future import annotations

import sys from pathlib import Path

sys.path.insert(0, str(Path(file).resolve().parents[1] / “src”))

import pandas as pd import streamlit as st

from chan_stock.data import store from chan_stock.engine import analyzer from chan_stock.settings import DEFAULT_LEVELS, FREQ_15M, FREQ_60M, FREQ_DAY, freq_label

st.title(“个股缠论分析”)

col_code, col_level, col_force = st.columns([2, 2, 1]) with col_code: code = st.text_input(“股票代码”, value=“600519”, help=“6 位 A 股代码”) with col_level: levels = st.multiselect( “分析级别”, [FREQ_DAY, FREQ_60M, FREQ_15M], default=DEFAULT_LEVELS, format_func=freq_label, ) with col_force: st.write(“”) force = st.checkbox(“强制刷新”, value=False)

col_watch, col_run = st.columns([1, 1]) with col_watch: if code and code.isdigit() and len(code) == 6: if store.is_watched(code): st.success(“该股已在自选股中”) elif st.button(“加入自选股”): store.add_watch(code) st.rerun() with col_run: run = st.button(“开始分析”, type=“primary”, use_container_width=True)

def _render_signals(result: dict) -> None: for freq in result[“levels”]: level_result = result[“results”][freq] sig = level_result[“signals”] st.subheader(f”{freq_label(freq)} 级别”) st.image(level_result[“chart_path”], use_container_width=True) c1, c2, c3 = st.columns(3) c1.metric(“买卖点数量”, sig[“bsp_count”]) c2.metric(“中枢数量”, sig[“zs_count”]) c3.metric(“最新价”, f”{sig[‘last_price’]:.2f}” if sig[“last_price”] else “-”)

    if sig["bsp"]:
        rows = [
            {
                "时间": p["time"],
                "信号": ("买点 " if p["is_buy"] else "卖点 ") + p["type"],
                "价格": p["price"],
            }
            for p in sig["bsp"][-10:]
        ]
        st.caption("最近买卖点")
        st.dataframe(pd.DataFrame(rows[::-1]), use_container_width=True)

    if sig["zs"]:
        zs_rows = [
            {"低点": z["low"], "高点": z["high"], "已确认": "是" if z["is_sure"] else "否"}
            for z in sig["zs"][-5:]
        ]
        st.caption("最近中枢区间")
        st.dataframe(pd.DataFrame(zs_rows[::-1]), use_container_width=True)

if run: if not code or not code.isdigit() or len(code) != 6: st.error(“请输入 6 位数字股票代码”) elif not levels: st.warning(“请至少选择一个分析级别”) else: try: with st.spinner(“正在拉取行情并计算缠论（首次拉取可能较慢）…”): result = analyzer.analyze(code, levels=levels, force_refresh=force) _render_signals(result) except Exception as e: # noqa: BLE001 st.error(f“分析失败：{e}”) pages/watchlist.py “”“自选股页面：添加 / 移除 / 一键扫描实时行情。”“” from future import annotations

import sys from pathlib import Path

sys.path.insert(0, str(Path(file).resolve().parents[1] / “src”))

import pandas as pd import streamlit as st

from chan_stock.data import feed, store

st.title(“自选股”)

col_add_code, col_add_btn = st.columns([3, 1]) with col_add_code: new_code = st.text_input(“添加股票代码”, value=“”, help=“6 位 A 股代码”) with col_add_btn: st.write(“”) if st.button(“添加”, use_container_width=True): if new_code and new_code.isdigit() and len(new_code) == 6: store.add_watch(new_code) st.rerun() else: st.warning(“请输入 6 位数字代码”)

items = store.list_watch() if not items: st.info(“暂无自选股。可在上方添加，或在「个股分析」页点击「加入自选股」。”) st.stop()

if st.button(“一键扫描实时行情”, type=“primary”): with st.spinner(“正在扫描实时行情…”): st.session_state[“scan_result”] = feed.get_realtime([it[“code”] for it in items])

if “scan_result” in st.session_state and st.session_state[“scan_result”]: rt = st.session_state[“scan_result”] rows = [] for it in items: c = it[“code”] q = rt.get(c, {}) name = q.get(“name”) or it.get(“name”) or “-” if q.get(“name”): store.update_watch_name(c, q[“name”]) rows.append( { “代码”: c, “名称”: name, “现价”: q.get(“price”), “涨跌幅%”: q.get(“pct”), } ) df = pd.DataFrame(rows)

def _color_pct(v):
    if v is None or pd.isna(v):
        return ""
    if v > 0:
        return "color: red"
    if v < 0:
        return "color: green"
    return ""

styled = df.style.map(_color_pct, subset=["涨跌幅%"])
st.dataframe(styled, use_container_width=True)

else: st.caption(“点击上方按钮扫描自选股实时行情。”)

st.divider() st.subheader(“自选股列表”) for it in items: c1, c2 = st.columns([4, 1]) c1.write(f”{it[‘code’]}　{it.get(‘name’) or ‘’}“) if c2.button(“移除”, key=f“rm_{it[‘code’]}”): store.remove_watch(it[“code”]) st.session_state.pop(“scan_result”, None) st.rerun() tests/test_config.py from future import annotations

import sys from pathlib import Path

sys.path.insert(0, str(Path(file).resolve().parents[1] / “src”))

from chan_stock.config import AppConfig

def test_default_config() -> None: config = AppConfig() assert config.app_name src/init.py “”“src 包入口。”“” src/chan_stock/init.py “”“项目包入口。”“” src/chan_stock/config.py from future import annotations

import os from dataclasses import dataclass from pathlib import Path import sys

from dotenv import load_dotenv

def _get_runtime_root() -> Path: if getattr(sys, “frozen”, False): return Path(sys.executable).resolve().parent return Path(file).resolve().parents[2]

def _load_runtime_env() -> None: runtime_root = _get_runtime_root() for candidate in (runtime_root / “.env”, runtime_root / “.env.example”): if candidate.exists(): load_dotenv(candidate, override=False) break

_load_runtime_env()

@dataclass(slots=True) class AppConfig: app_name: str = os.getenv(“APP_NAME”, “demo-app”) output_dir: str = os.getenv(“OUTPUT_DIR”, “output”) log_level: str = os.getenv(“LOG_LEVEL”, “INFO”) src/chan_stock/engine/init.py