我写了一个能查股票行情的小工具，真的很上头！

最近听粉丝留言，我折腾了一个小项目：A股行情查询与分析工具。（文末附源码）

先放上运行效果👇

✨ 功能概览

整个工具由四个核心文件组成：

• stock_fetcher.py：核心数据获取，封装了 Baostock + Akshare 双数据源；
• utils.py：通用工具函数（格式化、汇总、交易状态判断等）；
• config.py：全局配置（常量、显示规则等）；
• app.py：主入口，调用查询并展示结果。

实现的主要功能包括：

• 查询多只股票行情（最多 10 个）
• 自动兜底：Baostock 不可用时切换 Akshare
• 计算 1~10 日累计收益率
• 格式化展示：金额带“万/亿”，涨跌幅带正负号和颜色

🔍 技术解析

1. 多数据源兜底机制

核心逻辑在 get_stock_complete_data：

def get_stock_complete_data(self, stock_code: str) -> Optional[Dict]:
    if self.baostock_logged_in:
        data = self._fetch_with_baostock(stock_code)
        if data:
            return data
    return self._fetch_with_akshare(stock_code)

• 优先尝试 Baostock（数据更全）
• 如果失败，自动回退到 Akshare
• 避免因为单一接口挂掉而导致整个程序不可用

👉 这是一种 高可用设计，相当于给行情接口加了“双保险”。

2. 频率控制与请求重试

行情接口容易被刷爆，所以加了 _rate_limit 和 requests 的重试策略：

retry_strategy = Retry(
    total=max_retries,
    status_forcelist=[429, 500, 502, 503, 504],
    backoff_factor=1
)

• 限速：两次请求之间间隔 0.5s；
• 失败重试：遇到 429/500 等错误时自动重试；
• 退避策略：每次重试间隔逐渐延长，避免压垮接口。

👉 这保证了工具能长期稳定运行。

3. 累计收益率的计算

在 calculate_cumulative_returns 里：

cumulative_return = ((end_price / start_price) - 1) * 100

它会计算 过去 1~10 日的累计收益率。

• 如果你在写量化策略，可以直接用它来做“短期趋势筛选”；
• 这一步其实就是小型“策略引擎”的雏形。

4. 数据格式化与可读性优化

在 utils.py 里：

• 金额自动转“万/亿”
• 百分比自动加正负号
• 涨跌幅配红绿样式

def format_currency(amount: float) -> str:
    if abs(amount) >= 1e8:
        return f"{amount/1e8:.2f}亿"
    elif abs(amount) >= 1e4:
        return f"{amount/1e4:.2f}万"
    return f"{amount:.2f}"

👉 数据可读性大幅提升，避免一长串没意义的大数字。

5. 整体市场温度计

不仅能查单只股票，还能统计市场整体表现：

metrics = {
    'total_stocks': len(data_list),
    'positive_count': len([x for x in changes if x > 0]),
    'avg_change': np.mean(changes),
    'total_amount': sum(amounts),
}

最终可以得到：

• 市场上涨家数 vs 下跌家数
• 平均涨跌幅
• 成交总额、市值规模

👉 相当于做了一个迷你版的 市场热力图。

✅ 总结

这个项目总结：

• 多数据源兜底 → 保证可用性
• 限速与重试机制 → 保证稳定性
• 格式化展示 → 提升可读性
• 市场整体指标 → 从查询走向分析

📌 附源码

项目由5个文件组成，完整项目可以直接复制到本地运行，运行streamlit run app.py –server.port 8501即可。