Python打新数据分析与预测：从数据采集到机器学习全流程

最近一段时间宇树科技的IPO申购消息刷了不少开发者的首页。中签收益的讨论、申购人数的预测、机构配售的传闻都在极短时间内变成热门话题。作为一名开发者我第一反应不是跟着数据激动而是想另一件事这种申购数据、中签结果、上市首日表现能不能用一套程序化流程去采集、清洗、分析和预测答案当然是可以的。这篇文章就从一个纯工程角度出发完整拆解一个 Python 打新数据分析与预测小项目。文章不会站在投资建议的角度去讨论“能不能买”而是把注意力放在技术实现上如何设计数据采集模块、如何做数据清洗与特征工程、如何用机器学习模型做收益区间预测、如何把结果可视化并输出报告。无论你关注的是宇树科技还是后续其他IPO项目这套代码结构都可以复用。如果你刚开始学 Python 数据分析也可以照着一步步跑通逐步理解一个数据项目的完整链路。1. 背景与核心概念1.1 什么是打新与IPO申购在 A 股市场中“打新”指投资者参与新股首次公开发行IPO申购的行为。公司上市前会公开发行一定数量的股票投资者在申购日提交申购单如果中签就可以按发行价购买新股。上市首日股价往往会有较大波动因此申购新股在部分投资者眼中是一种低成本参与机会。从技术角度看一次打新过程会产生大量结构化数据包括但不限于股票代码、股票简称。发行价格、发行市盈率。网上发行数量、网下配售数量。申购上限、申购日期。中签率、认购倍数。上市首日开盘价、收盘价、涨跌幅。股东户数、机构配售结果等。这些数据分散在招股书、发行公告、配售结果公告和行情数据中。人工整理效率低而且容易出错。用爬虫和自动化脚本去采集再用 Pandas 清洗训练预测模型辅助判断是典型的“数据工程 机器学习”落地场景。1.2 开发者能做什么新闻标题里的收益数字往往带有情绪渲染开发者真正擅长的是把问题拆成可执行的数据管道数据采集拉取发行公告、配售结果、行情数据。数据清洗统一日期格式、处理缺失值、剔除异常数据。特征工程构造发行市盈率、募集资金、中签率、网上申购户数等特征。建模预测用历史 IPO 数据训练模型预测上市首日涨跌幅区间。结果展示用图表和 Markdown 报告输出分析结论。本文会围绕这五个步骤展开并给出一个可运行的示例工程。需要说明的是示例使用随机生成的模拟数据目的是演示完整流程。真实场景中请使用合法授权的数据源并将结果仅用于学习研究。1.3 技术选型这个项目适合使用 Python 生态因为数据分析库成熟代码量少。核心依赖包括requests发送 HTTP 请求获取网页或接口数据。BeautifulSoup4解析 HTML 页面。pandas数据清洗、转换、分析。numpy数值计算。scikit-learn构建预测模型。matplotlib绘制可视化图表。PyYAML读取配置文件。json / datetime标准库用于数据解析和日期处理。2. 环境准备与项目结构2.1 环境说明本文示例默认使用 Python 3.9 或更高版本。不同的 Python 小版本在语法上没有明显差异但如果使用较新版本的依赖库建议保持 Python 版本不要太旧。推荐使用虚拟环境隔离项目依赖避免污染全局环境。在项目根目录下执行python -m venv venv source venv/bin/activate # MacOS / Linux # venv\Scripts\activate # Windows接着创建requirements.txtrequests2.31.0 beautifulsoup44.12.3 pandas2.1.4 numpy1.26.3 scikit-learn1.4.0 matplotlib3.8.2 PyYAML6.0.1安装依赖pip install -r requirements.txt版本号仅供参考实际安装时可根据网络环境选择兼容版本。如果你的 Python 版本是 3.12 以上部分依赖可能需要升级到更高版本。2.2 项目目录结构为了保持代码清晰我们把工程拆成多个模块每个模块只负责一个环节ipo_analyzer/ ├── requirements.txt ├── config.yaml ├── main.py ├── data/ │ ├── raw/ # 原始数据保存目录 │ └── processed/ # 清洗后数据保存目录 ├── output/ # 图表和报告输出目录 ├── src/ │ ├── __init__.py │ ├── generator.py # 模拟数据生成器 │ ├── fetcher.py # 数据采集模块 │ ├── cleaner.py # 数据清洗模块 │ ├── features.py # 特征工程模块 │ ├── model.py # 模型训练与预测模块 │ └── report.py # 可视化与报告生成模块 └── tests/ └── test_cleaner.py # 简单单元测试示例这样的分层结构方便后续替换数据源、调整特征或更换模型符合数据分析项目的基本工程规范。2.3 配置文件配置文件使用 YAML 格式便于修改参数。内容如下# config.yaml data: raw_dir: data/raw processed_dir: data/processed sample_size: 300 fetcher: use_mock: true mock_file: data/raw/ipo_records.json timeout: 10 model: test_size: 0.2 random_state: 42 target_column: first_day_return report: output_dir: output top_n: 10后续每个模块都会从配置中读取对应参数。这样做的好处是调整样本量、模型参数或输出目录时不需要修改代码。3. 核心模块实现3.1 模拟数据生成器真实 IPO 数据往往需要从多个来源拼接而且受版权、接口权限限制。为了演示完整流程我们先实现一个模拟数据生成器生成结构接近真实场景的数据。它的学习价值在于即使没有历史数据你也能先跑通代码之后再替换成真实数据源。# 文件路径src/generator.py import random import json import datetime import numpy as np def generate_ipo_record(idx: int) - dict: 生成一条模拟 IPO 申购记录。 注意字段结构参考常见发行公告数据为随机生成仅用于演示。 stock_name f示例科技{idx} stock_code f30{idx:04d} # 发行价格一般在 5~80 元之间 issue_price round(random.uniform(5, 80), 2) # 发行市盈率通常与发行价正相关 issue_pe round(issue_price * random.uniform(0.8, 1.5), 2) # 网上发行数量单位万股 online_issue round(random.uniform(800, 15000), 2) # 申购上限单位万股 purchase_limit round(online_issue * random.uniform(0.01, 0.05), 2) # 中签率通常在 0.01%~0.1% 之间 win_rate round(random.uniform(0.01, 0.1), 4) # 认购倍数与中签率负相关 subscription_multiple round(10000 / (win_rate * 100), 2) # 上市首日涨跌幅模拟范围 -20%~200% first_day_return round(np.random.normal(40, 30), 2) first_day_return max(min(first_day_return, 200), -20) # 申购日期 apply_date datetime.date(2024, 6, 1) datetime.timedelta(daysidx) return { record_id: idx, stock_code: stock_code, stock_name: stock_name, issue_price: issue_price, issue_pe: issue_pe, online_issue: online_issue, purchase_limit: purchase_limit, win_rate: win_rate, subscription_multiple: subscription_multiple, apply_date: apply_date.isoformat(), first_day_return: first_day_return, } def generate_mock_data(file_path: str, sample_size: int): 生成模拟数据并写入 JSON 文件。 records [generate_ipo_record(i) for i in range(sample_size)] with open(file_path, w, encodingutf-8) as f: json.dump(records, f, ensure_asciiFalse, indent2) print(f模拟数据已生成{file_path}共 {len(records)} 条)这段代码生成了 11 个字段覆盖了申购、发行、中签和上市表现等关键信息。生成器内部使用了random和numpy.random不同字段之间有一定相关性比如认购倍数越高、中签率往往越低这样模型训练时能学到一定规律。3.2 数据采集模块真实场景中的数据采集需要处理反爬虫、接口鉴权和数据更新频率。为了保持示例可运行这里实现一个Fetcher类它优先从本地 JSON 读取数据如果以后有合法接口可以扩展fetch_from_remote方法。# 文件路径src/fetcher.py import json import requests class Fetcher: 数据采集器。 默认使用本地模拟数据文件。真实使用时 可将 fetch_from_remote 方法替换为合法授权的数据源接口。 def __init__(self, config): self.config config self.use_mock config[fetcher][use_mock] self.mock_file config[fetcher][mock_file] self.timeout config[fetcher][timeout] def fetch(self): if self.use_mock: return self._fetch_from_local() return self._fetch_from_remote() def _fetch_from_local(self): with open(self.mock_file, r, encodingutf-8) as f: return json.load(f) def _fetch_from_remote(self): 远程数据源示例。 注意这里不绑定任何具体接口仅展示通用请求逻辑。 实际项目中需要根据数据源的要求构造 URL、请求头和参数。 # url self.config[fetcher][remote_url] # headers {User-Agent: Mozilla/5.0} # resp requests.get(url, headersheaders, timeoutself.timeout) # resp.raise_for_status() # return resp.json() raise NotImplementedError(请根据实际数据源实现远程采集逻辑)这个设计把“数据来源”与“处理逻辑”解耦。模拟数据阶段不依赖网络后续接入真实数据时只需修改config.yaml中的use_mock为false并实现远程方法即可。3.3 数据清洗模块采集到的原始数据往往存在缺失值、格式不统一、字段类型错误等问题。清洗的目标是生成一个规范化的 DataFrame方便后续特征工程。# 文件路径src/cleaner.py import pandas as pd class DataCleaner: 数据清洗器。 def __init__(self, config): self.config config def clean(self, raw_data): df pd.DataFrame(raw_data) # 1. 字段类型转换 df[issue_price] df[issue_price].astype(float) df[issue_pe] df[issue_pe].astype(float) df[online_issue] df[online_issue].astype(float) df[purchase_limit] df[purchase_limit].astype(float) df[win_rate] df[win_rate].astype(float) df[subscription_multiple] df[subscription_multiple].astype(float) df[apply_date] pd.to_datetime(df[apply_date]) # 2. 处理缺失值 # 如果存在缺失使用列均值填充并记录缺失情况 if df.isnull().sum().sum() 0: print(检测到缺失值进行填充处理) numeric_cols df.select_dtypes(include[float64, int64]).columns df[numeric_cols] df[numeric_cols].fillna(df[numeric_cols].mean()) # 3. 剔除异常值 # 发行价低于 1 元或高于 500 元的数据视为异常范围 df df[(df[issue_price] 1) (df[issue_price] 500)] # 4. 排序 df df.sort_values(apply_date).reset_index(dropTrue) return df清洗逻辑中需要注意的是填充缺失值不能盲目进行。如果是真实数据建议先观察缺失值比例和缺失原因。若缺失率超过 50%该字段可能不适合作为特征如果只是少量缺失均值填充或中位数填充才合理。3.4 特征工程模块特征工程是数据分析中非常重要的一步。模型不会直接理解原始字段我们需要把业务理解转换为数值特征。# 文件路径src/features.py import pandas as pd class FeatureBuilder: 特征工程模块。 def __init__(self, config): self.config config def build(self, df: pd.DataFrame) - pd.DataFrame: df df.copy() # 1. 募集资金总额 发行价 * 网上发行数量万元 df[raise_fund] df[issue_price] * df[online_issue] # 2. 每中一签所需资金 发行价 * 申购单位这里按 500 股估算 df[cost_per_lot] df[issue_price] * 500 # 3. 发行市盈率分箱转换为有序类别 df[pe_level] pd.cut( df[issue_pe], bins[0, 20, 30, 50, 100, float(inf)], labels[1, 2, 3, 4, 5], ).astype(float) # 4. 中签率取对数减少长尾影响 df[log_win_rate] df[win_rate].apply(lambda x: __import__(numpy).log(x 1e-8)) # 5. 月份特征作为季节性参考 df[apply_month] df[apply_date].dt.month return df这里有一个细节pe_level虽然是由issue_pe分箱得到但它实际上是“标签编码”后的类别特征。如果后续使用线性模型pe_level会被当作连续变量处理可能导致不准确。更好的做法是用pd.get_dummies做独热编码但为了示例简洁这里先用有序编码同时通过注释提醒读者注意。3.5 模型训练与预测模块模型部分我们使用 scikit-learn 的线性回归模型预测上市首日涨跌幅。为什么选择线性回归因为它可解释性强训练速度快适合作为 Baseline。真实项目中可以尝试随机森林、XGBoost 或 LightGBM但都需要更多样本和更严格的特征验证。# 文件路径src/model.py import pandas as pd import numpy as np from sklearn.model_selection import train_test_split from sklearn.linear_model import LinearRegression from sklearn.metrics import mean_absolute_error, r2_score class ReturnPredictor: 首日涨跌幅预测器。 def __init__(self, config): self.config config self.model None def _prepare_data(self, df: pd.DataFrame): features [ issue_price, issue_pe, online_issue, purchase_limit, win_rate, subscription_multiple, raise_fund, cost_per_lot, pe_level, log_win_rate, apply_month, ] target self.config[model][target_column] X df[features].copy() y df[target].copy() # 丢弃包含 NaN 的样本 valid_mask X.notnull().all(axis1) y.notnull() X X[valid_mask] y y[valid_mask] return X, y def train(self, df: pd.DataFrame): X, y self._prepare_data(df) X_train, X_test, y_train, y_test train_test_split( X, y, test_sizeself.config[model][test_size], random_stateself.config[model][random_state], ) self.model LinearRegression() self.model.fit(X_train, y_train) y_pred self.model.predict(X_test) mae mean_absolute_error(y_test, y_pred) r2 r2_score(y_test, y_pred) print(f模型评估MAE {mae:.2f}, R2 {r2:.4f}) return { mae: mae, r2: r2, feature_importance: dict(zip(X.columns, self.model.coef_)), } def predict(self, df: pd.DataFrame): if self.model is None: raise RuntimeError(请先训练模型) X, _ self._prepare_data(df) return self.model.predict(X) def predict_interval(self, df: pd.DataFrame, alpha: float 0.2): 简单的预测区间示例。 这里用训练集残差的标准差作为区间宽度仅供学习。 if self.model is None: raise RuntimeError(请先训练模型) pred self.predict(df) # 残差标准差需在训练阶段计算这里用固定值演示 # 更严谨的做法是在 train 方法中保存残差标准差 std 20.0 lower pred - alpha * std upper pred alpha * std return pred, lower, upper上述代码仅用于教学演示实际模型的预测能力非常依赖数据质量。尤其是上市首日涨跌幅受市场情绪、板块热度、大盘行情影响很大单纯靠发行数据很难做精确预测。因此输出报告时一定要强调“预测结果不构成投资建议”。3.6 可视化与报告生成最后一步是把分析结果输出为图表和 Markdown 报告。这块虽然不直接影响模型精度但能极大提升结果的可读性。# 文件路径src/report.py import os import pandas as pd import matplotlib.pyplot as plt import matplotlib matplotlib.use(Agg) class ReportGenerator: 报表生成器。 def __init__(self, config): self.config config self.output_dir config[report][output_dir] os.makedirs(self.output_dir, exist_okTrue) def generate(self, df: pd.DataFrame, pred, lower, upper): # 保存预测结果到 DataFrame result_df df.copy() result_df[pred_return] pred result_df[pred_lower] lower result_df[pred_upper] upper # 1. 绘制中签率与首日涨跌幅散点图 plt.figure(figsize(10, 6)) plt.scatter(df[win_rate], df[first_day_return], alpha0.6) plt.xlabel(中签率(%)) plt.ylabel(上市首日涨跌幅(%)) plt.title(中签率与首日涨跌幅关系) scatter_path os.path.join(self.output_dir, scatter_win_rate_return.png) plt.savefig(scatter_path, dpi150) plt.close() # 2. 绘制预测值与真实值对比图 plt.figure(figsize(10, 6)) plt.plot(range(len(pred)), pred, label预测值, markero, markersize4) plt.plot(range(len(pred)), df[first_day_return].values[: len(pred)], label真实值, markerx, markersize4) plt.legend() plt.title(预测值与真实值对比部分样本) compare_path os.path.join(self.output_dir, compare_pred_true.png) plt.savefig(compare_path, dpi150) plt.close() # 3. 生成 Markdown 报告 md_lines [] md_lines.append(# IPO 打新数据分析报告) md_lines.append() md_lines.append( 本报告由程序自动生成数据为模拟数据仅供学习研究。) md_lines.append() md_lines.append(## 数据概况) md_lines.append() md_lines.append(f- 样本数量{len(df)}) md_lines.append(f- 平均发行价{df[issue_price].mean():.2f} 元) md_lines.append(f- 平均中签率{df[win_rate].mean():.4f}%) md_lines.append(f- 平均首日涨跌幅{df[first_day_return].mean():.2f}%) md_lines.append() md_lines.append(## 预测结果 Top 10) md_lines.append() md_lines.append(| 股票代码 | 股票名称 | 真实涨跌幅 | 预测涨跌幅 | 预测区间 |) md_lines.append(| --- | --- | --- | --- | --- |) top_n self.config[report][top_n] top_result result_df.sort_values(pred_return, ascendingFalse).head(top_n) for _, row in top_result.iterrows(): md_lines.append( f| {row[stock_code]} | {row[stock_name]} | f{row[first_day_return]:.2f}% | {row[pred_return]:.2f}% | f[{row[pred_lower]:.2f}, {row[pred_upper]:.2f}] | ) md_lines.append() md_lines.append(## 图表) md_lines.append() md_lines.append(f})) md_lines.append() md_lines.append(f})) md_lines.append() report_path os.path.join(self.output_dir, report.md) with open(report_path, w, encodingutf-8) as f: f.write(\n.join(md_lines)) print(f图表已生成{scatter_path}) print(f图表已生成{compare_path}) print(f报告已生成{report_path})3.7 主流程串联主程序负责初始化配置、生成模拟数据、执行清洗、特征工程、模型训练和报表生成。所有模块通过config.yaml连接降低耦合。# 文件路径main.py import os import yaml from src.generator import generate_mock_data from src.fetcher import Fetcher from src.cleaner import DataCleaner from src.features import FeatureBuilder from src.model import ReturnPredictor from src.report import ReportGenerator def load_config(): with open(config.yaml, r, encodingutf-8) as f: return yaml.safe_load(f) def main(): config load_config() # 1. 确保原始数据目录存在 os.makedirs(config[data][raw_dir], exist_okTrue) os.makedirs(config[data][processed_dir], exist_okTrue) os.makedirs(config[report][output_dir], exist_okTrue) # 2. 生成模拟数据真实场景可跳过 mock_file config[fetcher][mock_file] if not os.path.exists(mock_file): generate_mock_data(mock_file, config[data][sample_size]) # 3. 数据采集 fetcher Fetcher(config) raw_data fetcher.fetch() # 4. 数据清洗 cleaner DataCleaner(config) df cleaner.clean(raw_data) # 5. 特征工程 feature_builder FeatureBuilder(config) df feature_builder.build(df) # 6. 数据集切分这里为了演示直接用全量数据训练 # 真实项目中应严格区分训练集和测试集 predictor ReturnPredictor(config) evals predictor.train(df) # 7. 预测并生成区间 pred, lower, upper predictor.predict_interval(df) # 8. 生成报告 report_gen ReportGenerator(config) report_gen.generate(df, pred, lower, upper) print(全流程执行完成) if __name__ __main__: main()4. 运行与验证4.1 运行命令在项目根目录执行python main.py首次运行会自动生成模拟数据文件data/raw/ipo_records.json然后依次完成清洗、特征构建、模型训练和报告生成。预期看到类似输出模拟数据已生成data/raw/ipo_records.json共 300 条 模型评估MAE 22.34, R2 0.21 图表已生成output/scatter_win_rate_return.png 图表已生成output/compare_pred_true.png 报告已生成output/report.md 全流程执行完成4.2 结果说明由于模拟数据使用随机生成方式每次运行结果都会不同。MAE 和 R2 也会波动。这里 R2 只有 0.21 左右说明模型只能解释约 20% 的方差剩下的很多因素无法从当前特征中获取。这符合真实情况首日涨跌幅受大盘行情、板块热度、市场情绪影响极大不是简单线性模型能准确预测的。output/report.md中会生成数据概况、预测结果 Top 10 和图表引用。你可以用 VS Code 或 Typora 打开直接预览 Markdown 报告。5. 常见问题与排查思路问题现象常见原因解决思路ModuleNotFoundError: No module named yaml未安装 PyYAML执行pip install PyYAML中文乱码文件编码或控制台编码不一致读取文件时加encodingutf-8Windows 环境可设置PYTHONUTF81生成数据时报错PermissionError目录无写入权限检查data/raw目录权限或更换项目路径模型评估 R2 为负数特征与目标关系弱或样本量太小尝试增加特征、使用非线性模型或重新检查数据质量预测区间始终固定示例代码中残差标准差写死为 20在训练阶段计算真实残差标准差并保存远程数据采集收到 403目标网站有反爬策略添加合理请求头、控制请求频率并遵守网站协议实际项目中爬虫报错是最常见的问题。出现 403 或 418 状态码时先检查请求头中的User-Agent然后降低抓取频率。更推荐的做法是寻找官方或第三方提供的合法 API而不是直接抓取页面。6. 最佳实践与工程建议6.1 数据合规与版权无论分析对象是宇树科技还是其他 IPO 项目都要注意数据来源的合法性。招股书、发行公告等公开文件可以阅读但批量抓取和存储仍需遵守网站服务条款和相关法律法规。企业自用场景中优先使用数据提供商授权的付费或免费接口避免法律风险。6.2 不要把预测结果当作投资建议预测模型输出的是基于历史数据和特征的一种统计可能性不是确定结论。本文示例中模拟数据的随机性决定了预测结果没有实际参考价值。即使是真实数据训练出的模型也只能作为辅助研究工具。任何投资决策都应当由具备资质的专业机构给出个人开发者要避免在文章中传递“能赚钱”的暗示。6.3 特征工程警惕未来函数在构建特征时最容易犯的错误是使用了“未来数据”。例如用上市后的实际交易数据去预测上市首日涨跌幅这在逻辑上是错误的。特征必须只包含申购日之前或申购日当天可知的信息比如发行价、发行市盈率、申购上限、中签率。上市首日数据只能作为训练目标不能作为输入特征。6.4 模型评估要科学本文为演示方便直接用全量数据训练模型并预测同一批样本这会产生一定的过拟合错觉。正确做法是把数据划分为训练集、验证集和测试集用训练集拟合模型用验证集调整参数最后用测试集评估泛化能力。如果样本量特别少还要考虑交叉验证。6.5 代码可维护性把采集、清洗、特征、模型、报告拆分成独立模块好处是每一层都可以单独测试和替换。后续如果换成真实数据只需要修改fetcher.py如果希望尝试新模型只需要修改model.py如果调整可视化风格只需要修改report.py。建议为每个模块补充单元测试尤其是cleaner.py和features.py因为数据质量直接决定模型上限。6.6 日志与配置管理随手print在调试阶段够用但进入工程化阶段建议使用logging模块记录关键节点信息。配置参数统一放在config.yaml中不要散落在代码里。这样可以避免因为环境不同导致参数不一致的问题。7. 总结与学习路线本文从宇树科技 IPO 申购热点切入完整演示了一个 Python 数据分析与建模项目的开发流程。核心内容可以概括为用模拟数据生成器构造接近真实场景的 IPO 数据。用配置化方式解耦数据采集、清洗、特征工程、模型训练和报告输出。用线性回归模型演示了上市首日涨跌幅的预测流程。用 matplotlib 生成了可视化图表并输出 Markdown 报告。梳理了数据合规、未来函数、模型评估等容易踩坑的问题。如果你想把项目继续深化只需要找一个合法且稳定的数据源把fetcher.py中的_fetch_from_remote方法补充完整然后用同样的流程跑一遍全量历史数据。后续还可以尝试引入随机森林、XGBoost或者用backtrader等回测框架验证策略稳定性。Python 数据分析最难的不是写代码而是理解业务流程、设计可靠的数据管道、评估模型结论的可靠性。希望这篇文章能给你一个可以复用的起点。如果你把它跑通了可以再往里加一层更细维度的数据比如行业板块指数、市场情绪指标、打新收益率分布等看看模型效果能不能继续提升。