3个步骤手写实现信息消费概念股分析工具
刚学完Python语法，是不是对着IDE发呆？知道怎么写if-else和for循环，但面对“信息消费概念股”这种复杂业务需求，脑子一片空白。别慌，今天带你手写实现一个完整的概念板块分析项目，把零散语法串成可用工程。
项目目标与场景定义
很多人以为“信息消费概念股”只是股市里的一个标签，但在工程落地时，它背后是数据采集、清洗、指标计算、可视化的完整链路。
我们的目标很明确：
- 输入：一组股票代码 + 所属概念板块映射关系
- 处理：拉取历史行情、计算动量指标、筛选高波动个股
- 输出：生成一份包含趋势图与排名表的HTML报告
这不是简单调库，而是让你理解：每个API背后在做什么，数据从哪来，异常怎么兜底。
目录结构设计
先搭骨架，再填肉。项目结构如下：
info-concept-analyzer/
├── config/
│ └── settings.py # 全局配置：API密钥、超时、重试次数
├── core/
│ ├── data_fetcher.py # 数据获取层
│ ├── indicator.py # 指标计算层
│ └── reporter.py # 报告生成层
├── utils/
│ └── logger.py # 日志工具
├── main.py # 入口文件
└── requirements.txt # 依赖清单
为什么这样分？ 因为当你把“拿数据”“算指标”“出报告”拆开，每个模块可独立测试、独立替换。比如明天想换数据源，只改data_fetcher.py，其他层不动。
核心代码实现
1. 配置层：把“魔法数字”关进笼子
config/settings.py
# config/settings.py
import os# 从环境变量读取，避免硬编码密钥
TIANQI_API_KEY = os.getenv("TIANQI_API_KEY", "your_key_here")
REQUEST_TIMEOUT = 10 # 秒
MAX_RETRIES = 3 # 失败重试次数
CONCEPT_NAME = "信息消费概念股" # 目标概念板块
STOCK_POOL = ["002230", "600588", "002049", "300033"] # 示例股票池
关键点：永远不要在生产代码里写死API密钥。用环境变量是底线，上生产还得接密钥管理服务。
2. 数据获取层：带重试的HTTP请求
core/data_fetcher.py
# core/data_fetcher.py
import requests
import time
import logging
from config.settings import TIANQI_API_KEY, REQUEST_TIMEOUT, MAX_RETRIESlogger = logging.getLogger(__name__)class DataFetcher:def __init__(self, api_key: str):self.api_key = api_keyself.base_url = "https://api.example.com/v1"def get_kline(self, stock_code: str, start_date: str, end_date: str) -> list:"""获取日线K线数据，内置指数退避重试"""url = f"{self.base_url}/kline"params = {"symbol": stock_code,"start": start_date,"end": end_date,"token": self.api_key}for attempt in range(MAX_RETRIES):try:resp = requests.get(url, params=params, timeout=REQUEST_TIMEOUT)resp.raise_for_status()data = resp.json().get("data", [])logger.info(f"[{stock_code}] 成功获取 {len(data)} 条K线")return dataexcept requests.exceptions.RequestException as e:wait_time = 2 ** attempt # 1s, 2s, 4slogger.warning(f"[{stock_code}] 第{attempt+1}次请求失败: {e}, {wait_time}s后重试")time.sleep(wait_time)logger.error(f"[{stock_code}] 重试{MAX_RETRIES}次后仍失败，返回空列表")return []
逐行拆解：
raise_for_status()：HTTP 4xx/5xx 会抛异常，不写这行，错误会被静默吞掉。2 ** attempt：指数退避，避免服务器压力大时雪崩。- 最终返回空列表而非抛异常：上层业务可以优雅降级，而不是整个进程崩溃。
3. 指标计算层：手写动量指标
core/indicator.py
# core/indicator.py
import pandas as pddef calc_momentum(df: pd.DataFrame, window: int = 20) -> pd.Series:"""计算N日动量指标：(当前收盘价 - N日前收盘价) / N日前收盘价手动实现，不依赖talib，理解原理比调库重要"""if df.empty:return pd.Series(dtype=float)close = df['close'].values# 构造N日前收盘价序列，前N-1个位置填充NaNprev_close = pd.Series(close).shift(window).values# 避免除零with pd.option_context('mode.use_inf_as_na', True):momentum = (close - prev_close) / prev_closereturn pd.Series(momentum, index=df.index, name='momentum')def rank_by_momentum(data_dict: dict, top_n: int = 5) -> pd.DataFrame:"""按动量值排序，返回Top Ndata_dict: {stock_code: pd.Series(momentum)}"""records = []for code, series in data_dict.items():latest_mom = series.dropna().iloc[-1] if not series.dropna().empty else 0.0records.append({"code": code, "momentum": round(latest_mom, 4)})df = pd.DataFrame(records).sort_values('momentum', ascending=False)return df.head(top_n)
为什么手写？ 很多教程直接
import talib，但你不知道动量本质就是价格变化率。手写一遍，面试被问“动量指标公式是什么”才能脱口而出。
4. 报告生成层：生成HTML
core/reporter.py
# core/reporter.py
from datetime import datetimedef generate_html_report(top_df: pd.DataFrame, concept: str) -> str:"""生成简单HTML报告，含表格与生成时间"""now = datetime.now().strftime("%Y-%m-%d %H:%M")table_rows = ""for _, row in top_df.iterrows():table_rows += f"<tr><td>{row['code']}</td><td>{row['momentum']}</td></tr>\n"html = f"""<html><head><title>{concept}分析报告</title></head><body><h1>{concept} 动量排名 Top {len(top_df)}</h1><p>生成时间: {now}</p><table border="1"><thead><tr><th>股票代码</th><th>动量值</th></tr></thead><tbody>{table_rows}</tbody></table><p>数据来源: 示例API | 实现方式: 手写Python</p></body></html>"""return html
5. 主流程串联
main.py
# main.py
import logging
from config.settings import STOCK_POOL, CONCEPT_NAME
from core.data_fetcher import DataFetcher
from core.indicator import calc_momentum, rank_by_momentum
from core.reporter import generate_html_report# 配置日志
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
logger = logging.getLogger("main")def main():fetcher = DataFetcher(api_key="demo_key")momentum_dict = {}for code in STOCK_POOL:kline = fetcher.get_kline(code, "2023-01-01", "2023-12-31")if not kline:logger.warning(f"{code} 无数据，跳过")continue# 转换为DataFrameimport pandas as pddf = pd.DataFrame(kline)df['close'] = df['close'].astype(float)df = df.sort_values('date') # 确保时间有序mom_series = calc_momentum(df, window=20)momentum_dict[code] = mom_seriesif not momentum_dict:logger.error("所有股票均无有效数据，退出")returntop_df = rank_by_momentum(momentum_dict, top_n=5)html = generate_html_report(top_df, CONCEPT_NAME)output_path = "report.html"with open(output_path, "w", encoding="utf-8") as f:f.write(html)logger.info(f"报告已生成: {output_path}")print(top_df.to_string(index=False))if __name__ == "__main__":main()
运行与测试
环境准备
pip install requests pandas
本地运行
python main.py
预期输出：
2024-06-15 10:23:01 [INFO] main: 报告已生成: report.htmlcode momentum
002230 0.1523
600588 0.0987
002049 -0.0234
300033 -0.0512
打开report.html，能看到带边框的表格和生成时间戳。
单元测试建议
至少为calc_momentum写测试：
# tests/test_indicator.py
import pandas as pd
from core.indicator import calc_momentumdef test_momentum_basic():df = pd.DataFrame({'close': [10, 11, 12, 13, 14],'date': ['2023-01-01', '2023-01-02', '2023-01-03', '2023-01-04', '2023-01-05']})mom = calc_momentum(df, window=2)# 第3天: (12-10)/10 = 0.2assert abs(mom.iloc[2] - 0.2) < 1e-6
优化扩展方向
1. 并发获取数据
当前是串行请求，100只股票要等很久。用concurrent.futures改造：
from concurrent.futures import ThreadPoolExecutor, as_completeddef fetch_all_stocks(fetcher, stock_list, start, end):results = {}with ThreadPoolExecutor(max_workers=5) as executor:futures = {executor.submit(fetcher.get_kline, code, start, end): codefor code in stock_list}for future in as_completed(futures):code = futures[future]try:results[code] = future.result()except Exception as e:logger.error(f"{code} 异常: {e}")return results
2. 数据持久化
把K线数据存SQLite，避免重复请求：
import sqlite3def save_kline_to_db(db_path, stock_code, df):conn = sqlite3.connect(db_path)df.to_sql('kline', conn, if_exists='replace', index=False,schema={"symbol": stock_code})conn.close()
3. 可视化升级
接入plotly生成交互式图表，替换静态HTML表格。
4. 部署为API
用FastAPI包装main()逻辑，暴露/analyze接口，前端调用即可。
小结与互动
这个项目没有用任何“高级框架”，全是requests+pandas+标准库，但手写实现的过程让你彻底搞懂：
- HTTP请求的生命周期与异常处理
- 指标计算的数值细节与边界情况
- 模块解耦的工程思维
- 从0到1搭建可运行项目的完整路径
很多初学者卡在“语法会了，项目不会搭”，本质是缺少一个完整闭环的练习。这个项目可以直接放到GitHub 开源仓库里，作为你技术栈的“活简历”——面试官问“你做过什么项目”，你能说出数据流、异常处理、性能优化的细节，比背十本教程都有用。
还有什么不懂的？评论区留言挨个回。 比如“怎么加缓存”“怎么接真实数据源”“怎么部署到云服务器”，直接问，我一条条拆给你看。
