Python自动化股票持仓查询：从API调用到定时任务部署实战

1. 项目概述：为什么散户需要自动化查询

如果你还在每天手动打开券商APP，一个个数着股票盈亏，然后打开Excel表格手动记录，那你的时间可能正被大量重复劳动所消耗。对于散户而言，信息获取的及时性和准确性，是做出交易决策的基础。然而，人工操作不仅效率低下，还容易出错。想象一下，你持仓了十几只股票，每天收盘后要花半小时整理数据，一周就是两个多小时，这些时间本可以用来研究公司财报或市场趋势。

“炒股自动化”的核心，第一步就是实现资产和持仓数据的自动查询。这不仅仅是省时间，更是将投资行为从“体力活”升级为“技术活”的关键跳板。通过Python调用券商或数据服务商提供的API（应用程序编程接口），你可以让程序在每天收盘后自动拉取你的账户总资产、持仓明细、成本价、市价、浮动盈亏等关键数据，并自动保存到数据库或生成可视化报表。这样，你就能从繁琐的数据搬运工角色中解放出来，专注于策略思考和决策本身。

这个项目适合所有对Python有基础了解，并希望提升自己投资管理效率的散户。你不需要是编程高手，只需要会基础的Python语法，并愿意花一点时间理解API的工作逻辑。接下来，我将以一个典型的流程为例，手把手带你走通从环境准备、接口申请、代码编写到数据处理的完整路径，并分享我在这过程中踩过的坑和总结的技巧。

2. 核心思路与方案选型：不走弯路的架构设计

在动手写代码之前，理清思路和选对方案至关重要。盲目开始很容易陷入“代码能跑，但不好用、不安全、不可靠”的困境。我的核心设计思路遵循三个原则：安全性第一、稳定性优先、可扩展性预留。

2.1 数据源的选择：券商API vs 第三方数据平台

这是第一个关键决策点，直接决定了后续所有工作的走向。

1. 券商官方API：

   • 优点：数据最权威、最实时。查询的是你本人证券账户的真实数据，包含精确的成本、持仓、可用资金等。部分券商还支持模拟交易、条件单等高级功能。
   • 缺点：门槛较高。大型券商（如华泰、中信、国泰君安等）通常只为机构客户或量化私募提供API服务，对散户不开放或申请流程复杂。即使开放，也需要临柜办理、签署协议，且可能有资金门槛。文档和支持可能不完善。
   • 适用场景：资金量较大、交易频繁，且券商支持API服务的资深散户。

2. 第三方金融数据平台API：

   • 优点：接入方便，文档齐全。像Wind、Tushare、AkShare、JoinQuant（聚宽）等平台提供了丰富的金融市场数据API，部分平台通过模拟账户或与券商合作，也能提供个人账户的查询功能（需授权）。它们通常有完善的Python SDK和社区支持。
   • 缺点：可能涉及数据权限和费用。查询真实持仓需要你将券商账户授权给平台，存在一定的隐私和安全顾虑。部分高级数据或实时数据需要付费。
   • 适用场景：绝大多数散户入门学习的首选。可以先从免费的数据接口（如查询公开行情）开始，再逐步过渡到需要账户授权的持仓查询。

我的选择与建议：对于初学者和大多数散户，我强烈建议从第三方平台开始。例如，Tushare Pro或AkShare提供了相对友好的入门方式。你可以先用它们来获取行情数据，理解API调用的整个流程。等整个自动化框架搭建成熟后，再考虑是否要攻克券商官方API。本篇文章的后续示例，也将以第三方数据平台的模式进行讲解，因为它更具普适性。

2.2 技术栈的确定：轻量、高效、易维护

我们的目标是构建一个轻量级的自动化查询工具，而不是一个庞大的量化交易系统。因此，技术栈要精简。

• 核心语言：Python。这是金融数据分析领域的事实标准，库生态丰富。
• 网络请求库：requests。简单易用，足以应对绝大多数HTTP API的调用。
• 数据处理库：pandas。查询回来的数据（通常是JSON格式）用pandas的DataFrame进行处理、分析和保存，事半功倍。
• 数据存储：初期可以使用CSV文件或SQLite数据库，轻便无需额外安装。后期数据量大可考虑MySQL或PostgreSQL。
• 定时任务：使用系统自带的crontab（Linux/macOS）或任务计划程序（Windows）来定时执行Python脚本。这是最简单稳定的方案，无需引入额外的Python调度库。
• 配置文件：使用config.ini或config.yaml文件来管理API密钥、账户信息等敏感配置，绝对不要硬编码在代码中。

这个技术栈组合，确保了项目易于上手、运行稳定，并且每个环节都有成熟的社区支持。

3. 实战准备：从零搭建你的自动化环境

理论清晰后，我们开始动手。这里我以使用一个假设的、类似Tushare的第三方数据平台“FinData API”为例，因为它涵盖了通用API调用的所有核心环节。

3.1 环境搭建与依赖安装

首先，确保你的电脑安装了Python（建议3.8及以上版本）。打开终端（或命令提示符），创建一个专属的项目目录，并安装必要的库。

# 创建项目目录并进入 mkdir stock_auto_query cd stock_auto_query # 创建虚拟环境（推荐，避免包冲突） python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/macOS: source venv/bin/activate # 安装核心库 pip install requests pandas

虚拟环境能隔离项目依赖，是Python项目开发的好习惯。

3.2 获取并安全配置API凭证

这是整个项目的安全命脉。我们前往“FinData”平台官网注册账号，通常会在“个人中心”或“API管理”页面找到你的API凭证，一般包括：

• api_token：用于身份验证的唯一令牌。
• api_base_url：API服务的基础地址，如https://api.findata.com/v1。

关键安全操作：创建一个名为config.ini的配置文件来保存它们。

; config.ini [API] base_url = https://api.findata.com/v1 token = your_actual_api_token_here # 请替换成你自己的token [ACCOUNT] # 如果是模拟账户或已授权的账户ID account_id = your_account_id

然后在代码中读取这个配置，并确保将config.ini添加到.gitignore文件中，防止误提交到公开仓库导致密钥泄露。

# config_loader.py import configparser import os def load_config(): config = configparser.ConfigParser() config.read('config.ini') # 确保config.ini文件在当前目录或指定路径 return config # 使用示例 cfg = load_config() API_BASE_URL = cfg['API']['base_url'] API_TOKEN = cfg['API']['token'] ACCOUNT_ID = cfg['ACCOUNT']['account_id']

4. 核心代码实现：一步步构建查询引擎

环境就绪，密钥备好，现在我们来编写最核心的API调用与数据处理代码。

4.1 构建通用的API请求函数

一个健壮的请求函数需要处理认证、错误和重试。我们将其封装起来，方便所有查询调用。

# api_client.py import requests import pandas as pd import time from config_loader import load_config cfg = load_config() API_BASE_URL = cfg['API']['base_url'] API_TOKEN = cfg['API']['token'] HEADERS = { 'Authorization': f'Token {API_TOKEN}', 'Content-Type': 'application/json' } def make_api_request(endpoint, params=None, method='GET', max_retries=3): """ 发送API请求的通用函数 :param endpoint: API端点路径，如 '/account/assets' :param params: 请求参数（字典） :param method: 请求方法，GET或POST :param max_retries: 最大重试次数 :return: 请求成功的JSON数据，或抛出异常 """ url = f"{API_BASE_URL}{endpoint}" for attempt in range(max_retries): try: if method.upper() == 'GET': response = requests.get(url, headers=HEADERS, params=params, timeout=10) else: response = requests.post(url, headers=HEADERS, json=params, timeout=10) # 检查HTTP状态码 response.raise_for_status() # 非200状态码会抛出HTTPError # 解析JSON响应 data = response.json() # 检查API业务逻辑是否成功（假设成功时返回的JSON包含 `code: 0`） if data.get('code') != 0: raise Exception(f"API业务错误: {data.get('msg', 'Unknown error')}") return data.get('data') # 返回数据部分 except requests.exceptions.RequestException as e: print(f"第{attempt+1}次网络请求失败: {e}") if attempt < max_retries - 1: wait_time = 2 ** attempt # 指数退避 print(f"{wait_time}秒后重试...") time.sleep(wait_time) else: raise Exception(f"请求失败，已达最大重试次数: {url}") except ValueError as e: raise Exception(f"响应JSON解析失败: {e}")

这个函数做了几件重要的事：添加认证头、处理网络异常、实现指数退避重试、检查HTTP状态和业务状态码。这是生产级代码的雏形。

4.2 查询账户总资产

有了通用请求函数，查询资产就变得非常简单。我们需要知道对应的API端点（Endpoint）和参数。

# query_assets.py from api_client import make_api_request from config_loader import load_config cfg = load_config() ACCOUNT_ID = cfg['ACCOUNT']['account_id'] def query_account_assets(): """ 查询账户总资产信息 :return: 包含总资产、可用资金、总市值等的字典 """ endpoint = '/account/assets' params = { 'account_id': ACCOUNT_ID } try: asset_data = make_api_request(endpoint, params=params) # 假设返回的数据结构如下： # { # "total_asset": 1000000.00, // 总资产 # "available_cash": 150000.00, // 可用资金 # "market_value": 850000.00, // 持仓市值 # "frozen_cash": 0.00 // 冻结资金 # } print("账户资产查询成功！") print(f"总资产: {asset_data.get('total_asset'):.2f} 元") print(f"可用资金: {asset_data.get('available_cash'):.2f} 元") print(f"持仓市值: {asset_data.get('market_value'):.2f} 元") return asset_data except Exception as e: print(f"查询账户资产失败: {e}") return None if __name__ == "__main__": assets = query_account_assets()

4.3 查询详细持仓列表

持仓查询通常返回一个列表，每条记录代表一只股票或基金的持仓情况。用pandas处理这种表格数据再合适不过。

# query_positions.py from api_client import make_api_request from config_loader import load_config import pandas as pd cfg = load_config() ACCOUNT_ID = cfg['ACCOUNT']['account_id'] def query_account_positions(): """ 查询账户持仓明细 :return: 包含所有持仓记录的pandas DataFrame """ endpoint = '/account/positions' params = { 'account_id': ACCOUNT_ID } try: positions_data = make_api_request(endpoint, params=params) # 假设返回的数据是一个列表，每个元素是一只股票的持仓信息 # [ # { # "symbol": "000001.SZ", // 股票代码 # "symbol_name": "平安银行", // 股票名称 # "current_amount": 1000, // 当前持仓数量 # "available_amount": 1000, // 可卖数量 # "cost_price": 15.50, // 成本价 # "market_price": 16.20, // 当前市价 # "market_value": 16200.00, // 持仓市值 # "float_profit_loss": 700.00, // 浮动盈亏 # "profit_loss_ratio": 0.0452 // 盈亏比例 # }, # ... // 其他持仓 # ] if positions_data: # 转换为DataFrame df_positions = pd.DataFrame(positions_data) # 计算一些衍生字段（如果API未提供） if 'cost_price' in df_positions.columns and 'market_price' in df_positions.columns: df_positions['cost_value'] = df_positions['current_amount'] * df_positions['cost_price'] print("持仓查询成功！") print(f"共持有 {len(df_positions)} 只标的。") # 打印一个简明的持仓概览 print(df_positions[['symbol_name', 'current_amount', 'market_price', 'market_value', 'float_profit_loss']].to_string(index=False)) return df_positions else: print("持仓列表为空。") return pd.DataFrame() # 返回空DataFrame except Exception as e: print(f"查询持仓失败: {e}") return pd.DataFrame() if __name__ == "__main__": df = query_account_positions() # 可以在这里将df保存为CSV或写入数据库 if not df.empty: df.to_csv('daily_positions.csv', index=False, encoding='utf-8-sig') print("持仓数据已保存至 daily_positions.csv")

4.4 数据持久化与简单分析

查询到数据不是终点，自动保存和历史分析才是自动化的价值所在。我们可以创建一个主脚本，将查询和保存逻辑整合，并加入简单的分析。

# main.py import sys import os sys.path.append(os.path.dirname(__file__)) from query_assets import query_account_assets from query_positions import query_account_positions import pandas as pd from datetime import datetime import sqlite3 def save_to_csv(asset_data, position_df, date_str=None): """将当日数据保存到CSV文件""" if date_str is None: date_str = datetime.now().strftime('%Y-%m-%d') # 保存资产快照 if asset_data: asset_df = pd.DataFrame([asset_data]) asset_df['date'] = date_str asset_file = f'data/assets_{date_str}.csv' asset_df.to_csv(asset_file, index=False, encoding='utf-8-sig') print(f"资产数据已保存至 {asset_file}") # 保存持仓快照 if not position_df.empty: position_df['date'] = date_str position_file = f'data/positions_{date_str}.csv' position_df.to_csv(position_file, index=False, encoding='utf-8-sig') print(f"持仓数据已保存至 {position_file}") def save_to_sqlite(asset_data, position_df, date_str=None): """将数据保存到SQLite数据库，便于历史查询和分析""" if date_str is None: date_str = datetime.now().strftime('%Y-%m-%d') conn = sqlite3.connect('portfolio.db') # 保存资产记录 if asset_data: asset_data['date'] = date_str asset_df = pd.DataFrame([asset_data]) asset_df.to_sql('account_assets', conn, if_exists='append', index=False) # 保存持仓记录 if not position_df.empty: position_df['date'] = date_str position_df.to_sql('account_positions', conn, if_exists='append', index=False) conn.close() print(f"数据已存入SQLite数据库 (portfolio.db)") def generate_daily_report(position_df): """生成简单的当日持仓报告""" if position_df.empty: print("今日无持仓，无需生成报告。") return total_mv = position_df['market_value'].sum() total_pl = position_df['float_profit_loss'].sum() print("\n========== 当日持仓报告 ==========") print(f"持仓总市值: {total_mv:.2f} 元") print(f"持仓总浮动盈亏: {total_pl:.2f} 元") print(f"持仓标的数量: {len(position_df)}") # 找出盈亏最多的三只股票 top_gainers = position_df.nlargest(3, 'float_profit_loss') top_losers = position_df.nsmallest(3, 'float_profit_loss') print("\n【盈利前三】") for _, row in top_gainers.iterrows(): print(f" {row['symbol_name']}({row['symbol']}): 盈利 {row['float_profit_loss']:.2f} 元") print("\n【亏损前三】") for _, row in top_losers.iterrows(): print(f" {row['symbol_name']}({row['symbol']}): 亏损 {abs(row['float_profit_loss']):.2f} 元") print("==================================\n") if __name__ == "__main__": # 确保数据目录存在 os.makedirs('data', exist_ok=True) print(f"开始执行自动化查询任务 @ {datetime.now()}") # 1. 查询资产 print("\n[步骤1] 查询账户总资产...") asset_info = query_account_assets() # 2. 查询持仓 print("\n[步骤2] 查询账户持仓明细...") positions_df = query_account_positions() # 3. 生成报告 print("\n[步骤3] 生成日报...") generate_daily_report(positions_df) # 4. 保存数据 print("\n[步骤4] 持久化数据...") today_str = datetime.now().strftime('%Y%m%d') save_to_csv(asset_info, positions_df, today_str) save_to_sqlite(asset_info, positions_df, today_str) print("\n自动化查询任务完成！")

5. 部署与自动化：让脚本自己定时运行

代码在本地跑通只是成功了一半，让它在收盘后自动运行，才是真正的“自动化”。

5.1 使用系统定时任务（以Linux/macOS的crontab为例）

这是最经典、最稳定的方法。假设你的主脚本路径是/home/yourname/stock_auto_query/main.py。

1. 打开crontab编辑界面：

   crontab -e

2. 在文件末尾添加一行，设定每天下午15:30（A股收盘后）执行：

   30 15 * * 1-5 cd /home/yourname/stock_auto_query && /home/yourname/stock_auto_query/venv/bin/python main.py >> /home/yourname/stock_auto_query/cron.log 2>&1

   • 30 15 * * 1-5：表示周一到周五（1-5）的15点30分。
   • cd ...：切换到项目目录。
   • venv/bin/python：使用虚拟环境中的Python解释器。
   • main.py：要执行的脚本。
   • >> cron.log 2>&1：将脚本的标准输出和错误输出都重定向到cron.log文件，方便日后排查问题。

5.2 Windows任务计划程序

对于Windows用户，可以通过图形界面设置。

1. 搜索并打开“任务计划程序”。
2. 点击“创建基本任务”。
3. 按照向导，设置任务名称、触发器（每天、工作日）、开始时间（15:30）。
4. 在“操作”步骤，选择“启动程序”，程序或脚本填写你的Python解释器全路径（如C:\Users\YourName\stock_auto_query\venv\Scripts\python.exe），参数填写main.py的全路径，起始于填写项目目录。
5. 完成创建。

5.3 进阶：添加简单的异常通知

自动化运行后，我们还需要知道它是否成功。一个简单的方法是让脚本在失败时给自己发一封邮件。

# notifier.py import smtplib from email.mime.text import MIMEText from email.header import Header import traceback def send_error_email(subject, error_msg): """发送错误通知邮件（需预先配置发件邮箱）""" # 这里需要你配置自己的邮箱SMTP信息 mail_host = "smtp.163.com" # 例如163邮箱SMTP服务器 mail_user = "your_email@163.com" mail_pass = "your_authorization_code" # 注意是授权码，不是登录密码 sender = mail_user receivers = ['your_notification_email@example.com'] # 接收邮件的地址 message = MIMEText(error_msg, 'plain', 'utf-8') message['From'] = Header("Stock Auto Query Bot", 'utf-8') message['To'] = Header("管理员", 'utf-8') message['Subject'] = Header(f"[自动化脚本异常] {subject}", 'utf-8') try: smtp_obj = smtplib.SMTP_SSL(mail_host, 465) # 163邮箱SSL端口 smtp_obj.login(mail_user, mail_pass) smtp_obj.sendmail(sender, receivers, message.as_string()) print("错误邮件发送成功") except Exception as e: print(f"发送错误邮件失败: {e}") finally: try: smtp_obj.quit() except: pass # 在主脚本main.py的异常捕获块中调用 # try: # ... 你的主要逻辑 ... # except Exception as e: # error_msg = f"任务执行失败: {str(e)}\n\n{traceback.format_exc()}" # send_error_email("每日持仓查询任务失败", error_msg) # raise # 可以选择重新抛出异常，让crontab记录日志

6. 避坑指南与常见问题排查

在实际操作中，你几乎一定会遇到下面这些问题。我把我的踩坑经验总结在这里，希望能帮你节省大量时间。

6.1 API调用常见错误与解决

6.2 数据与存储的坑

• 时间戳问题：API返回的时间可能是Unix时间戳（10位或13位整数）或特定格式的字符串。用pd.to_datetime()转换时，务必明确指定单位（unit='s'或unit='ms'）或格式（format='%Y-%m-%d %H:%M:%S'）。
• 浮点数精度：金融计算涉及小数，使用Python的float类型可能会产生精度误差。对于精确计算（如成本价*数量），建议使用Decimal类型。但在大多数展示和报表场景下，float并保留两位小数即可。
• CSV文件乱码：用Excel打开CSV出现乱码时，在to_csv()方法中指定encoding='utf-8-sig'参数可以解决。
• 数据库连接未关闭：如果使用SQLite或MySQL，每次操作完务必conn.close()，或者使用with上下文管理器，避免程序长时间运行后连接泄露。

6.3 安全与维护要点

• 密钥管理是红线：再次强调，config.ini必须加入.gitignore。可以考虑使用环境变量来存储密钥（如os.getenv('API_TOKEN')），这样更安全。
• 日志记录必不可少：无论是crontab的重定向，还是在代码中使用logging模块，都必须有日志。当脚本无声无息失败时，日志是唯一的救命稻草。
• 定期检查与更新：第三方平台的API可能会升级，接口地址或字段可能变化。每隔一段时间，运行一下脚本，确认功能正常。订阅平台的公告频道也是个好习惯。
• 功能边界清晰：我们这个脚本的核心是“查询”，不要让它承担过多的计算或分析逻辑，保持单一职责。复杂的分析可以交给另一个专门的分析脚本，通过读取数据库或CSV文件来进行。

走到这里，你已经拥有了一个每天自动为你查询资产和持仓的“数字助理”。它安静、可靠、准确，将你从重复劳动中彻底解放。但这仅仅是炒股自动化的起点。基于这个稳定的数据流，你可以轻松地扩展出更多功能：自动计算每日收益率、绘制资产曲线图、监控特定股票的股价提醒、甚至对接钉钉/企业微信机器人推送日报。当你把基础的数据获取管道搭建牢固后，上层的各种应用想象空间才会被真正打开。