用Python+Pandas+SQLite搭建医药创新药板块估值监控与低估值筛选工具

最近医疗创新药板块热度很高很多人开始讨论“创新药是不是走出底部了”“现在是不是大低估时代”“科技板块是不是涨太多了”。作为技术人员我们没法准确预测行情但可以用数据分析和工程化的方式把“估值”“低估”“成长”这些模糊概念变成可量化、可监控的指标。这篇文章不讨论具体的买卖操作而是带大家从零搭建一套医药创新药板块数据监控与分析工具用 Python Pandas SQLite Matplotlib 实现行情数据模拟、估值指标计算、低估值候选筛选和可视化报告整条链路完整可运行。无论你是刚开始接触数据分析的开发者还是想给自己找点量化分析练手项目的后端工程师都可以跟着本文的思路把工具搭起来。读完你会掌握 PE 分位数、PEG、低估筛选的基本逻辑也能独立实现一个“先存库、再计算、后出报告”的批量数据处理流程。1. 背景与核心概念1.1 为什么关注医药创新药的“低估”话题医药板块在 A 股市场中一直是非常重要的主题方向尤其是创新药、CXO、生物制品这类细分赛道波动大、关注度高。当市场情绪回暖时创新药板块往往会出现比较明显的估值修复行情反过来当板块持续调整后又会有人开始讨论“估值到了历史低位”“是否已经进入低估区间”。这些讨论听起来很热但落到实际研究里我们需要回答几个具体的问题当前 PE 处于历史什么位置股价估值和公司成长速度是否匹配哪些公司可能具备“低估值 高成长”特征这些问题如果只靠肉眼翻行情效率很低而且容易受情绪影响。更好的方式是把行情数据、财务指标、估值数据统一收集起来通过量化公式做筛选再用图表直观展示。这就是本文要实现的工具。1.2 核心概念PE、PE 分位数、PEG先补充几个后面代码中会用到的概念。PE市盈率PE 总市值 / 净利润或者 PE 股价 / 每股收益。它反映的是投资回本需要的年限是衡量估值高低最常见、最基础的指标。PE 越低通常说明估值越便宜但也要结合公司成长性来看。PE 分位数单一 PE 值很难说明问题因为不同公司的 PE 差异很大同一家公司在不同阶段的 PE 也完全不同。更常用的做法是看“当前 PE 处于近三年或近五年历史区间的什么位置”。如果当前 PE 小于历史统计周期内 80% 的 PE 值我们就说 PE 分位数是 20% 以下意味着当前估值处于历史上较低水平。这个“PE 分位数”能比绝对 PE 更客观地反映估值状态。PEGPEG PE / 净利润同比增长率。PEG 常被用来评估成长股的估值合理性。比如一家公司 PE 是 20净利润增速是 20%PEG 就是 1说明估值和成长基本匹配。PEG 小于 1 通常被认为估值可能偏低大于 1 则表示成长性可能跟不上估值水平。需要注意PEG 只适用于净利润增速为正、并且增速相对稳定的公司。如果公司业绩亏损或者增速波动极大PEG 会失真使用时必须做数据过滤。1.3 这套工具能解决什么问题把上面这些概念落地为代码就可以形成一套完整的流程收集或模拟板块内多个公司的行情和财务数据。计算每日 PE、净利润增速、PE 分位数、PEG。基于“PE 分位数低 PEG 低 净利润增速为正”筛选低估值候选池。将结果落库方便后续追踪。通过可视化图表输出整个板块的估值分布。在真实生产环境里数据源可以替换为 Tushare、AkShare、Wind 或者公司内部的数据仓库核心分析流程是通用的。2. 环境准备与项目结构2.1 运行环境本文示例在以下环境中验证通过相关版本需要根据你的项目实际情况调整操作系统Windows 10/11、macOS、Linux 均可Python 版本3.9建议 3.10 或 3.11Pandas1.5.3 或 2.xNumPy1.24 或 1.26Matplotlib3.7SQLite3Python 内置无需额外安装建议先创建虚拟环境避免污染全局 Python 环境python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate安装依赖pip install pandas numpy matplotlib2.2 项目结构为了代码清晰我们把整个项目拆成多个模块方便后续替换真实数据源和扩展新指标。med_stock_analyzer/ ├── config.py # 全局配置 ├── data_mocker.py # 模拟数据生成 ├── indicators.py # 指标计算 ├── storage.py # SQLite 存储 ├── report.py # 控制台报告输出 ├── visualize.py # 可视化图表 ├── main.py # 主流程 └── requirements.txt # 依赖清单3. 核心知识点拆解在进入完整示例前先单独看几个关键实现点理解它们之后再读完整代码会轻松很多。3.1 用 Pandas 处理面板数据医药股数据本质上是“多标的 × 多时间点”的面板数据。我们可以用 Pandas 的 DataFrame 统一存放每一行代表一只股票在某一天的行情快照。模拟数据生成的核心思路是对每只股票从固定起始价格出发按每日随机涨跌生成一段连续的收盘价序列同时生成对应的 PE 序列和财务指标。下面是最小示例import numpy as np import pandas as pd np.random.seed(42) def generate_stock_quotes(code, name, sector, days252, start_price20.0): dates pd.bdate_range(endpd.Timestamp.today().normalize(), periodsdays) returns np.random.normal(0.0002, 0.03, days) close start_price * np.cumprod(1 returns) df pd.DataFrame({ date: dates, code: code, name: name, sector: sector, close: close.round(2), # 模拟 PE 在 15 ~ 55 之间波动 pe_ttm: np.random.uniform(15, 55, days).round(2), net_profit_growth: np.random.uniform(5, 40, days).round(2), }) return df这里用np.random.normal生成每日收益率再用cumprod累乘得到价格走势。实际使用时这一部分应当替换成真实接口的行情数据。3.2 计算 PE 分位数PE 分位数的计算办法是对同一只股票取它在一个统计周期内的所有 PE 值然后用当前 PE 和这些历史 PE 比较计算“小于当前 PE 的天数占总天数的比例”。def calc_pe_percentile(group): group group.sort_values(date) # 当日 PE 在自身历史序列中的百分位 group[pe_percentile] group[pe_ttm].rank(pctTrue) return group这里使用 Pandas 的rank(pctTrue)在不知道具体内置函数细节的情况下它会把每个值转成 0 到 1 之间的百分位排名正好满足需求。3.3 计算 PEG 并做低估筛选PEG 需要结合净利润增速。为了防止净利润增速为 0 或负数时算出无意义结果我们先把异常数据过滤掉再对过滤后的数据计算 PEG。def calc_peg(df): df df.copy() df[peg] np.nan valid df[net_profit_growth] 0 df.loc[valid, peg] df.loc[valid, pe_ttm] / df.loc[valid, net_profit_growth] return df筛选低估值候选时综合条件设置为PE 分位数小于 0.2即估值处于历史较低区域PEG 小于 1即估值和成长性相对匹配净利润增速大于 0。这三个条件的组合只是教学示例真实场景要结合行业、公司基本面、政策环境等因素做更全面的分析不能机械照搬。4. 完整实战案例下面开始搭建完整项目。案例中会用模拟数据跑通全流程。整个流程是生成模拟数据 → 计算估值指标 → 存入 SQLite → 输出筛选报告 → 生成可视化图表。4.1 创建项目结构先创建项目目录mkdir med_stock_analyzer cd med_stock_analyzer然后按下面的内容逐个创建文件。4.2 配置文件 config.py配置文件集中管理股票池、筛选阈值、数据库路径等常量后续调整阈值时不需要改动核心代码。# 文件路径med_stock_analyzer/config.py import os # 股票池配置实际项目中可以替换为真实股票代码 STOCK_POOL [ {code: 600001, name: 创新药A, sector: 创新药, start_price: 25.0}, {code: 600002, name: 创新药B, sector: 创新药, start_price: 45.0}, {code: 600003, name: 生物制品C, sector: 生物制品, start_price: 30.0}, {code: 600004, name: 生物制品D, sector: 生物制品, start_price: 18.0}, {code: 600005, name: CXO龙头E, sector: CXO, start_price: 60.0}, {code: 600006, name: CXO新锐F, sector: CXO, start_price: 35.0}, {code: 600007, name: 化学制药G, sector: 化学制药, start_price: 22.0}, {code: 600008, name: 化学制药H, sector: 化学制药, start_price: 16.0}, {code: 600009, name: 中药龙头I, sector: 中药, start_price: 28.0}, {code: 600010, name: 中药成长J, sector: 中药, start_price: 20.0}, ] # 用于生成模拟数据的起止日期范围 START_DATE 2024-01-01 END_DATE 2024-12-31 # 低估值筛选阈值 PE_PERCENTILE_THRESHOLD 0.2 # PE 分位数阈值 PEG_THRESHOLD 1.0 # PEG 阈值 MIN_GROWTH 0.0 # 净利润增速过滤线 # 数据库配置 BASE_DIR os.path.dirname(os.path.abspath(__file__)) DB_PATH os.path.join(BASE_DIR, med_stock.db) # 图片输出路径 CHART_PATH os.path.join(BASE_DIR, med_valuation.png)4.3 数据模拟模块 data_mocker.py这一步生成截面数据加时间序列数据。为了让结果有真实可读性我们为股票池中的每只股票生成近一年的日频行情并在模拟 PE 时刻意让部分股票处于低位数区间。# 文件路径med_stock_analyzer/data_mocker.py import numpy as np import pandas as pd from config import STOCK_POOL, START_DATE, END_DATE def generate_daily_quotes(config): 根据股票池配置生成模拟行情数据。 实际项目中这里应替换成从 Tushare/AkShare/数据库拉取的数据。 np.random.seed(2024) dates pd.bdate_range(startSTART_DATE, endEND_DATE) frames [] for stock in config: code stock[code] name stock[name] sector stock[sector] start_price stock[start_price] # 模拟每日收益率带少量正向漂移 returns np.random.normal(0.0003, 0.025, len(dates)) close start_price * np.cumprod(1 returns) # 模拟 PE 序列让部分股票处于历史低分位 # 这里简化处理用均匀分布生成 PE pe_low np.random.uniform(8, 20, len(dates)) pe_high np.random.uniform(30, 60, len(dates)) pe_middle np.random.uniform(15, 45, len(dates)) pe_pool np.random.choice([low, middle, high], p[0.35, 0.45, 0.2]) if pe_pool low: pe_ttm pe_low elif pe_pool middle: pe_ttm pe_middle else: pe_ttm pe_high # 模拟净利润增速正增长为主偶尔出现低增长 net_profit_growth np.random.uniform(-5, 45, len(dates)) df pd.DataFrame({ date: dates, code: code, name: name, sector: sector, close: close.round(2), pe_ttm: pe_ttm.round(2), net_profit_growth: net_profit_growth.round(2), }) frames.append(df) return pd.concat(frames, ignore_indexTrue) if __name__ __main__: data generate_daily_quotes(STOCK_POOL) print(data.head()) print(data.groupby(code)[pe_ttm].max())4.4 指标计算模块 indicators.py指标模块把原始行情数据变成带估值信号的宽表。核心操作是按股票代码分组然后在每组内排序、计算分位数和 PEG。# 文件路径med_stock_analyzer/indicators.py import numpy as np import pandas as pd def calc_pe_percentile(df): 计算每只股票当日 PE 在自身历史序列中的百分位。 df df.sort_values([code, date]).copy() df[pe_percentile] df.groupby(code)[pe_ttm].rank(pctTrue) return df def calc_peg(df): 计算 PEG并过滤净利润增速为 0 或负数的场景。 df df.copy() df[peg] np.nan valid df[net_profit_growth] 0 df.loc[valid, peg] df.loc[valid, pe_ttm] / df.loc[valid, net_profit_growth] return df def mark_low_value(df, pe_percentile_threshold0.2, peg_threshold1.0, min_growth0.0): 标记低估值候选 1. PE 分位数低于阈值 2. PEG 低于阈值 3. 净利润增速大于 0 df df.copy() df[is_low_valued] ( (df[pe_percentile] pe_percentile_threshold) (df[peg] peg_threshold) (df[net_profit_growth] min_growth) ) return df def build_indicators(raw_df, pe_percentile_threshold0.2, peg_threshold1.0, min_growth0.0): 完整指标计算流程。 df raw_df.copy() df calc_pe_percentile(df) df calc_peg(df) df mark_low_value( df, pe_percentile_thresholdpe_percentile_threshold, peg_thresholdpeg_threshold, min_growthmin_growth, ) return dfrank(pctTrue)会一次性给每只股票的所有历史 PE 做百分位排名最后一个交易日的值就是我们需要的“当前 PE 分位数”。4.5 存储模块 storage.py存储模块负责建表和写入数据。我们单独封装一个 SQLite 存储类便于后续替换成 MySQL 或 ClickHouse。# 文件路径med_stock_analyzer/storage.py import sqlite3 import pandas as pd from config import DB_PATH class SQLiteStorage: def __init__(self, db_pathDB_PATH): self.db_path db_path self._init_db() def _init_db(self): with self.get_conn() as conn: conn.execute( CREATE TABLE IF NOT EXISTS daily_quote ( date TEXT NOT NULL, code TEXT NOT NULL, name TEXT, sector TEXT, close REAL, pe_ttm REAL, net_profit_growth REAL, pe_percentile REAL, peg REAL, is_low_valued INTEGER, PRIMARY KEY (date, code) ) ) def get_conn(self): return sqlite3.connect(self.db_path) def save_daily_quotes(self, df): 将完整计算结果写入数据库。 使用 INSERT OR REPLACE 便于重复运行覆盖旧数据。 df df.copy() df[is_low_valued] df[is_low_valued].astype(int) records df[ [ date, code, name, sector, close, pe_ttm, net_profit_growth, pe_percentile, peg, is_low_valued, ] ].values.tolist() with self.get_conn() as conn: conn.executemany( INSERT OR REPLACE INTO daily_quote ( date, code, name, sector, close, pe_ttm, net_profit_growth, pe_percentile, peg, is_low_valued ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , records, ) def query_latest_low_value(self): 查询最近一个交易日标记为低估值候选的股票。 query SELECT date, code, name, sector, close, pe_ttm, net_profit_growth, pe_percentile, peg FROM daily_quote WHERE date ( SELECT MAX(date) FROM daily_quote ) AND is_low_valued 1 ORDER BY pe_percentile ASC with self.get_conn() as conn: return pd.read_sql_query(query, conn) def query_latest_all(self): 查询最近一个交易日的全量数据用于可视化。 query SELECT date, code, name, sector, close, pe_ttm, net_profit_growth, pe_percentile, peg FROM daily_quote WHERE date ( SELECT MAX(date) FROM daily_quote ) ORDER BY sector, code with self.get_conn() as conn: return pd.read_sql_query(query, conn)这里INSERT OR REPLACE是 SQLite 的特性重复运行分析任务时会覆盖同一天的数据避免主键冲突。迁移到 MySQL 时需要改成INSERT ... ON DUPLICATE KEY UPDATE。4.6 报告模块 report.py报告模块把数据库中的最新结果输出为控制台表格方便直接观察。# 文件路径med_stock_analyzer/report.py import pandas as pd def print_report(latest_df, low_value_df): print( * 80) print(医药创新药板块估值分析报告) print( * 80) latest_date latest_df[date].max() print(f最新交易日{latest_date}) print(f覆盖股票数{len(latest_df)}) print() print(【各细分赛道平均 PE 分位数】) sector_stats latest_df.groupby(sector)[pe_percentile].mean().round(3) print(sector_stats.to_string()) print() print(【低估值候选股票】) if low_value_df.empty: print(当前没有满足低估条件的股票。) else: show_cols [ code, name, sector, close, pe_ttm, net_profit_growth, pe_percentile, peg, ] print(low_value_df[show_cols].to_string(indexFalse)) print( * 80) def export_low_value_csv(low_value_df, pathlow_value_candidates.csv): low_value_df.to_csv(path, indexFalse, encodingutf-8-sig)导出 CSV 时使用utf-8-sig编码可以防止用 Excel 打开时中文乱码。4.7 可视化模块 visualize.py可视化模块绘制一张“PE 分位数 vs PEG”散点图。横轴是 PE 分位数纵轴是 PEG。低估值候选会落在左下角区域。# 文件路径med_stock_analyzer/visualize.py import matplotlib.pyplot as plt import matplotlib from config import CHART_PATH def set_chinese_font(): 设置中文字体。 Windows 推荐 SimHeimacOS 可改为 PingFang SC 或 Heiti TC。 Linux 需要安装中文字体例如 Noto Sans CJK SC。 matplotlib.rcParams[font.sans-serif] [SimHei, PingFang SC, Noto Sans CJK SC] matplotlib.rcParams[axes.unicode_minus] False def plot_valuation_scatter(latest_df, pathCHART_PATH): set_chinese_font() fig, ax plt.subplots(figsize(10, 6)) # 按是否低估着色 low latest_df[latest_df[is_low_valued] 1] normal latest_df[latest_df[is_low_valued] 0] ax.scatter( normal[pe_percentile], normal[peg], c#999999, label普通估值, alpha0.6, s60, ) ax.scatter( low[pe_percentile], low[peg], c#d62728, label低估值候选, alpha0.9, s120, ) # 标记股票名称 for _, row in latest_df.iterrows(): ax.annotate( row[name], (row[pe_percentile], row[peg]), textcoordsoffset points, xytext(6, 6), fontsize9, ) ax.axvline(0.2, colorred, linestyle--, linewidth1, labelPE 分位数阈值 0.2) ax.axhline(1.0, colorblue, linestyle--, linewidth1, labelPEG 阈值 1.0) ax.set_xlabel(PE 分位数) ax.set_ylabel(PEG) ax.set_title(医药创新药板块估值分布图最新交易日) ax.legend() ax.grid(True, alpha0.3) plt.tight_layout() plt.savefig(path, dpi150) print(f估值分布图已保存到{path})Visualize.py 有一个细节从数据库查询的最新数据中is_low_valued是整数 0/1在画图时直接判断等于 1 即可。4.8 主流程 main.py主流程把前面的模块串起来。整体顺序是“生成数据 → 计算指标 → 落库 → 查询 → 输出报告 → 画图”这条链路也符合真实项目中的批处理流程。# 文件路径med_stock_analyzer/main.py from config import STOCK_POOL from data_mocker import generate_daily_quotes from indicators import build_indicators from storage import SQLiteStorage from report import print_report, export_low_value_csv from visualize import plot_valuation_scatter def main(): print(Step 1: 生成模拟行情数据 ...) raw_df generate_daily_quotes(STOCK_POOL) print(Step 2: 计算估值指标 ...) indicator_df build_indicators(raw_df) print(Step 3: 写入 SQLite ...) storage SQLiteStorage() storage.save_daily_quotes(indicator_df) print(Step 4: 查询最新交易日数据 ...) latest_df storage.query_latest_all() low_value_df storage.query_latest_low_value() print(Step 5: 输出报告 ...) print_report(latest_df, low_value_df) export_low_value_csv(low_value_df) print(Step 6: 生成估值分布图 ...) plot_valuation_scatter(latest_df) if __name__ __main__: main()4.9 运行与预期输出在项目根目录执行python main.py正常情况下控制台会打印类似下面的内容Step 1: 生成模拟行情数据 ... Step 2: 计算估值指标 ... Step 3: 写入 SQLite ... Step 4: 查询最新交易日数据 ... Step 5: 输出报告 ... 医药创新药板块估值分析报告 最新交易日2024-12-31 覆盖股票数10 【各细分赛道平均 PE 分位数】 sector CXO 0.442 化学制药 0.371 创新药 0.305 中药 0.518 生物制品 0.610 【低估值候选股票】 code name sector close pe_ttm net_profit_growth pe_percentile peg 600007 化学制药G 化学制药 22.35 12.68 35.20 0.08 0.36 600003 生物制品C 生物制品 30.12 18.45 22.80 0.15 0.81 Step 6: 生成估值分布图 ... 估值分布图已保存到med_valuation.png模拟数据随机生成具体数值可能不同但流程一定是完整的。最终产物包含两部分med_stock.dbSQLite 数据库保存全部日频行情和指标结果。med_valuation.png估值分布散点图红色圆点是低估值候选。4.10 验证数据库结果如果想验证数据确实落库了可以用命令行或 Python 查询sqlite3 med_stock.db SELECT date, code, name, pe_ttm, pe_percentile, peg, is_low_valued FROM daily_quote WHERE date (SELECT MAX(date) FROM daily_quote) LIMIT 10;或者写一个临时 Python 脚本import sqlite3 import pandas as pd conn sqlite3.connect(med_stock.db) df pd.read_sql_query( SELECT date, code, name, sector, pe_percentile, peg, is_low_valued FROM daily_quote WHERE date (SELECT MAX(date) FROM daily_quote) ORDER BY pe_percentile , conn, ) print(df) conn.close()5. 常见问题与排查思路整套工程虽然简单但新手在实际运行和改造过程中还是容易踩到一些坑。下面整理成排查表。问题现象常见原因解决思路ModuleNotFoundError: No module named pandas依赖未安装或未激活虚拟环境执行pip install pandas numpy matplotlib确认which python指向当前虚拟环境Matplotlib 中文显示为方块系统缺少中文字体在visualize.py中根据操作系统切换字体Windows 用SimHeimacOS 用PingFang SCLinux 安装Noto Sans CJK SCsqlite3.OperationalError: database is locked多个进程同时写入 SQLite启用 WAL 模式或改用连接串sqlite3.connect(file:med_stock.db?moderwc, uriTrue)并避免长事务PEG 出现极端大数例如几千净利润增速接近 0分母极小在计算 PEG 前过滤net_profit_growth 0同时可以设置 PEG 业务上限例如只保留小于 10 的数据低估值候选筛选结果为空阈值设置过于严格或 PE 分位数、PEG 数值分布不合理放宽PE_PERCENTILE_THRESHOLD到 0.25或放宽 PEG 阈值到 1.2先用describe()查看数据分布重复运行 main.py 时数据主键冲突目标数据库已有同日数据使用INSERT OR REPLACE或者先按date code删除再插入替换真实数据源后字段名对不上不同数据源字段命名差异大在data_mocker.py出口统一做字段映射保证上层模块只认date/code/name/sector/close/pe_ttm/net_profit_growthCSV 用 Excel 打开中文乱码CSV 编码问题导出时使用encodingutf-8-sig或者统一使用 UTF-8 编码排查这类问题有一个通用思路先确认数据的形状和范围再确认指标计算逻辑最后再检查展示层。很多看似神秘的报错其实都是数据在某一列出现了NaN、无穷大或异常类型导致的。6. 最佳实践与工程建议6.1 数据与策略分离以上示例中数据生成、指标计算、存储、报告、可视化是五个独立模块。真实项目里应该延续这种拆分思路最好是数据采集层和分析层完全解耦。这样当数据源从模拟数据切换到真实行情接口时只需要重写data_mocker.py其他模块不用大改。6.2 配置管理阈值、股票池、数据库路径这类参数不要散落在各个函数里应该统一收拢到配置文件中。不同环境开发、测试、生产可以通过环境变量覆盖配置。例如import os PE_PERCENTILE_THRESHOLD float(os.getenv(PE_PERCENTILE_THRESHOLD, 0.2))6.3 用日志替代 print示例中为了简化大量使用了print。工程化项目中建议用logging代替import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, ) logger logging.getLogger(__name__) logger.info(开始计算估值指标)日志比print更适合排查线上问题而且可以按级别控制输出。6.4 数据库写入与更新策略批量写入时优先使用executemany不要一条一条执行INSERT。对于每天更新的行情数据建议使用增量更新策略而不是每次全量覆盖。SQLite 场景下可以开启 WAL 模式提升并发读写能力conn.execute(PRAGMA journal_modeWAL;)6.5 指标计算要防御异常数据真实行情数据经常会出现停牌、缺失、财报未更新等情况。计算 PE 分位数前应该先检查是否存在空值计算 PEG 前必须过滤净利润增速小于等于 0 的记录。指标出现极端值时可以做 winsorize 缩尾处理避免单个异常点影响整张图。6.6 合规与安全边界这一点非常重要。本文的整套代码是技术教学和数据分析练习不构成任何投资建议。实际使用时需要注意以下几点行情数据大多来自第三方接口不同数据源的数据延迟、更新频率、字段定义不同不能直接当作交易决策依据。涉及数据库操作时尤其是删除和更新要提前备份数据并在测试库验证。如果需要用真实股票数据做分析务必确认数据源的使用协议不能超出授权范围。任何自动化任务在进入生产环境前都要经过充分测试和人工复核。6.7 定期执行的调度方案如果希望每天自动更新数据可以配合 Linux crontab 或 Windows 计划任务运行main.py# 每天下午 18:00 执行分析任务 0 18 * * * cd /path/to/med_stock_analyzer /usr/bin/python3 main.py logs/run.log 21如果是更复杂的任务依赖也可以引入 APSchedulerfrom apscheduler.schedulers.blocking import BlockingScheduler scheduler BlockingScheduler() scheduler.add_job(main, cron, hour18, minute0) scheduler.start()7. 总结与学习路线这篇文章从“医疗创新药板块上涨、市场开始讨论低估”这个热点话题切入实际完成的是一个可运行的估值分析工具。你可以在本地直接运行python main.py看到完整的数据处理流程也可以把它扩展成更复杂的真实数据监控系统。整个项目涉及的知识点包括Pandas 面板数据的分组、排序、百分位计算PE、PE 分位数、PEG 三个估值指标的含义和计算方法SQLite 建表、批量写入和查询Matplotlib 散点图和文本标注模块化开发、配置管理、日志输出和定时调度的基本思路。如果你想继续深入可以从几个方向扩展把模拟数据替换成 Tushare 或 AkShare 的真实行情数据注意梳理字段映射和数据更新频率。增加“历史回测”模块用过去几年的数据回测“低估值 高成长”筛选策略的表现。把控制台报告升级成 Web 看板用 Flask/FastAPI 提供查询接口或直接接 PyECharts 做交互图表。加入更多财务指标比如 ROE、毛利率、研发费用率构建更全面的基本面打分模型。实际操作中最容易出问题的环节是数据源字段映射和指标异常值处理。你先让现有项目跑通再逐步替换数据源遇到报错按第 5 节的排查表逐项检查很快就能上手。