1. 项目概述一个普通人的纪律性投资实验作为一个在股市里摸爬滚打了好几年的普通散户我最大的敌人从来不是市场而是我自己。追涨杀跌、情绪化交易、被各种“消息”牵着鼻子走这些毛病我几乎全占。每次复盘看着那些因为一时冲动而错失的利润或造成的亏损都深感无力。我意识到问题的核心在于缺乏纪律。市场波动时人性的贪婪和恐惧会被无限放大而单靠意志力去对抗成功率实在太低。于是我开始思考能不能用技术手段来“管住自己的手”能不能把我的投资逻辑和纪律写成代码让一个没有感情的“AI小助手”来替我执行观察、分析和部分决策这就是“AI股票小助手”项目的初衷。它不是一个追求暴利、神乎其神的“圣杯”策略而是一个普通投资者用于建立纪律、辅助决策、记录市场观察的自动化工具。核心目标不是战胜市场而是战胜那个在市场里容易失控的自己。这个项目本质上是一个个人量化观察笔记系统。它利用Python结合基础的金融数据API、简单的数据分析库以及规则引擎将我个人的市场观察逻辑比如估值区间、技术指标状态、新闻情绪等代码化。当市场情况触发我预设的“纪律规则”时例如股价突破我设定的高估线应停止买入或跌入低估区间但趋势未稳应继续观察系统会通过通知提醒我甚至在某些严格定义的场景下执行模拟的“纪律性操作”如记录观察日志、生成观望提示从而避免我在情绪驱动下做出非理性决策。整个过程我都将其视为一场关于投资纪律的、可追溯、可复现的自我实验。2. 核心设计思路从主观观察到客观规则这个项目的设计核心在于将模糊、主观的个人投资“感觉”和“纪律”转化为清晰、客观、可执行的代码规则。这不仅仅是写几行获取股价的脚本而是一个系统的思考与拆解过程。2.1 纪律的代码化定义你的“交易宪法”首先我需要梳理自己的投资原则并将其具体化。这就像为自己的投资行为制定一部“宪法”所有操作都不能违背这些根本原则。我主要从以下几个维度入手仓位管理纪律这是生命线。我规定单只个股的仓位不超过总资金的20%任何情况下都不在下跌趋势中盲目补仓。在代码里这就体现为对持仓市值和可用资金的持续监控函数。买卖点纪律我采用一种简单的“区间观察法”。对于关注的股票我会根据历史市盈率PE、市净率PB和近期价格通道手动设定一个“理想买入区间”、“持有观察区”和“警惕高估区”。代码的任务就是持续监控股价处于哪个区间。风险触发纪律当个股单日跌幅超过-7%或自买入后回撤超过-15%时必须触发“强制复查”流程而不是假装没看见。代码需要实时计算这些回撤数据。信息处理纪律明确区分“事实”与“噪音”。代码只处理来自指定权威渠道的财报发布日期、分红公告等结构化信息并对社交媒体或新闻标题进行简单的关键词情绪分析如出现“暴涨”、“崩盘”、“重磅”等词汇进行标记提醒我这是“高情绪化信息”需谨慎对待。将这些文本纪律转化为逻辑判断是项目的第一步。例如“不在下跌趋势中盲目补仓”这一条就需要代码判断当前价格是否在N日均线之下近期是否连续下跌如果是则即使股价进入“理想买入区间”系统给出的建议也不是“买入”而是“进入观察区间但趋势未稳建议继续观望”。2.2 技术栈选型轻量、高效、可扩展作为一个个人项目技术栈的选择遵循“够用、好用、易维护”的原则。核心语言Python。这是毋庸置疑的选择。在金融数据分析领域Python拥有Pandas、NumPy这样行业标准的数据处理库以及TA-Lib技术指标库、yfinance雅虎财经数据接口需注意其稳定性等丰富的生态资源。代码编写直观社区支持强大非常适合快速原型开发。数据获取akshare 备用数据源。我主要使用国内的akshare库它提供了A股、港股、美股、宏观、新闻等非常全面的数据接口且免费。为了避免单一数据源故障我会为关键数据如每日行情配置一个备用源比如baostock需要登录或某些券商提供的免费Level-1行情API通常有频率限制。数据分析PandasNumPy。Pandas的DataFrame是处理时间序列数据的利器清洗数据、计算指标、区间统计都非常方便。NumPy则用于底层数值计算。规则引擎纯Python逻辑 schedule库。目前阶段我的规则还不复杂直接用if-elif-else逻辑链配合函数封装就能清晰表达。定时任务使用轻量的schedule库可以方便地设置“每个交易日开盘前”、“每30分钟”等定时检查任务。通知与日志logging 邮件/钉钉/微信机器人。所有操作、规则触发、市场状态变化都通过Python内置的logging模块记录到本地文件便于复盘。即时通知我选择了钉钉机器人因为它与手机钉钉绑定提醒非常及时。也可以使用smtplib发邮件或使用wechatpy等库接入微信。开发环境VSCode Jupyter Notebook。VSCode用于编写完整的脚本和模块Jupyter Notebook则用于数据探索、策略原型验证和可视化分析两者结合非常高效。注意yfinance等国外库在访问时可能存在网络不稳定问题。本项目作为个人纪律工具主要关注A股市场因此优先使用akshare等国内数据源完全避免了此类依赖也符合数据获取的便利性和稳定性要求。2.3 系统架构草图一个循环的观察者整个系统的运行模式可以看作一个“感知-判断-记录-提醒”的循环。数据感知层定时如每30分钟从akshare获取自选股列表的实时行情、基本面快照市盈率、市净率。同时可以定期如每日收盘后爬取或接入指定的财经新闻摘要。规则判断层这是核心。数据送入不同的“纪律检查器”仓位检查器计算当前总持仓、个股持仓占比。买卖点检查器根据预设的估值区间和当前股价判断股票所处状态。风险检查器计算日内涨跌幅、持仓回撤率。信息过滤器对获取的新闻标题进行简单的情感分析。决策记录层规则引擎输出结果。这里至关重要系统不直接执行真实交易而是生成“决策建议”和“纪律日志”。例如“腾讯控股当前股价进入‘警惕高估区’触发纪律规则禁止买入建议审视持仓理由。”“宁德时代日内跌幅达-8.2%触发强制复查规则。”输出提醒层将重要的纪律触发日志尤其是禁止性建议和风险警报通过钉钉机器人实时推送到手机。所有日志均写入本地文件数据库如SQLite或CSV文件形成可查询的“市场观察笔记”。这个架构的关键在于闭环我的每一次手动操作或放弃操作都应该能在系统日志中找到对应的纪律依据或提醒记录。长此以往这份“笔记”就成了我投资行为习惯的客观病历价值巨大。3. 关键模块实现与代码解析接下来我们深入到代码层面看看几个核心模块如何具体实现。我会使用akshare作为主要数据源因为它对A股支持非常友好。3.1 数据获取与清洗模块可靠的数据是系统的眼睛。我将其封装为一个独立的类DataFetcher。import akshare as ak import pandas as pd import numpy as np from datetime import datetime, timedelta import time import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class DataFetcher: def __init__(self): self.stock_list [sh.600519, sz.300750, sz.000858] # 示例贵州茅台宁德时代五粮液。akshare的代码格式 self.cache {} # 简单缓存避免短时间内重复请求 self.cache_ttl 60 # 缓存有效期秒 def get_realtime_quote(self, stock_code): 获取实时行情 cache_key frt_{stock_code} if cache_key in self.cache and (time.time() - self.cache[cache_key][timestamp]) self.cache_ttl: return self.cache[cache_key][data] try: # 使用akshare的stock_zh_a_spot_em接口获取实时数据 df ak.stock_zh_a_spot_em() # 该接口返回所有A股数据我们需要筛选出目标股票 # 注意akshare的实时接口代码格式可能为600519我们需要匹配 code_simple stock_code.split(.)[1] # 取出600519 stock_data df[df[代码] code_simple] if not stock_data.empty: # 提取关键字段 quote { code: stock_code, name: stock_data.iloc[0][名称], latest: stock_data.iloc[0][最新价], change: stock_data.iloc[0][涨跌幅], volume: stock_data.iloc[0][成交量], amount: stock_data.iloc[0][成交额], timestamp: datetime.now() } self.cache[cache_key] {data: quote, timestamp: time.time()} logger.info(f实时行情获取成功: {stock_code}) return quote else: logger.warning(f未找到股票实时数据: {stock_code}) return None except Exception as e: logger.error(f获取实时行情失败 {stock_code}: {e}) return None def get_hist_daily(self, stock_code, start_date20230101): 获取历史日K线数据用于计算指标和区间 cache_key fhist_{stock_code}_{start_date} if cache_key in self.cache: return self.cache[cache_key] try: # stock_zh_a_hist 获取历史行情 # 参数period可选daily, weekly, monthly df ak.stock_zh_a_hist(symbolstock_code.split(.)[1], perioddaily, start_datestart_date, adjustqfq) if df is not None and not df.empty: df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume }, inplaceTrue) df[date] pd.to_datetime(df[date]) df.set_index(date, inplaceTrue) self.cache[cache_key] df logger.info(f历史数据获取成功: {stock_code}, 共{len(df)}条) return df else: logger.warning(f未找到历史数据: {stock_code}) return None except Exception as e: logger.error(f获取历史数据失败 {stock_code}: {e}) return None def get_basic_info(self, stock_code): 获取基本面信息如PE、PB # 注意实时PE/PB数据接口可能不稳定或收费这里使用akshare的stock_a_lg_indicator最新指标作为示例 try: df ak.stock_a_lg_indicator(symbolstock_code.split(.)[1]) if df is not None and not df.empty: latest df.iloc[-1] # 取最新一条 info { pe_ttm: latest[市盈率], pb: latest[市净率], div_rate: latest[股息率] if 股息率 in latest else None } return info return None except Exception as e: logger.error(f获取基本面信息失败 {stock_code}: {e}) return None实操要点缓存机制实时行情接口频繁调用可能被封也浪费资源。添加一个简单的内存缓存能有效降低请求频率。异常处理网络请求必然存在失败可能必须用try-except包裹并通过日志记录避免程序因单次请求失败而崩溃。数据清洗akshare返回的数据字段名多为中文为了方便后续处理最好统一重命名为英文。日期列也建议转换为datetime类型并设为索引。3.2 纪律规则引擎模块这是项目的“大脑”。我将不同的纪律条款实现为独立的检查函数然后由一个主引擎来协调调度。class DisciplineEngine: def __init__(self, config): config: 字典包含各股票的纪律规则配置 示例配置 config { sh.600519: { buy_zone: {pe_max: 35, pe_min: 25, price_max: 1800, price_min: 1500}, sell_zone: {pe_min: 40, price_min: 2000}, position_limit: 0.2, # 仓位上限20% loss_alert: -0.07 # 日内跌幅超-7%警报 } } self.config config self.current_positions {} # 模拟或连接实盘获取的持仓 {‘code’: {‘shares’: 100, ‘cost_price’: 1600}} self.total_capital 1000000 # 模拟总资金 def check_position_limit(self, stock_code, planned_buy_value): 检查单票仓位是否超限 if stock_code not in self.config: return True, 未配置该股票规则 limit self.config[stock_code].get(position_limit, 0.2) current_value 0 if stock_code in self.current_positions: # 这里需要当前股价假设通过参数传入或实时获取 # current_value current_price * self.current_positions[stock_code][shares] pass # 简化处理 future_ratio (current_value planned_buy_value) / self.total_capital if future_ratio limit: return False, f计划买入将导致仓位({future_ratio:.1%})超过限制({limit:.0%})纪律禁止。 return True, 仓位检查通过 def check_valuation_zone(self, stock_code, current_pe, current_price): 检查估值区间 if stock_code not in self.config: return HOLD, 未配置规则 rules self.config[stock_code] buy_zone rules.get(buy_zone, {}) sell_zone rules.get(sell_zone, {}) zone HOLD # 默认持有观察区 reason [] # 买入区判断 (需同时满足PE和价格条件且趋势等其他因子在外部判断) if buy_zone: pe_ok buy_zone.get(pe_min, -np.inf) current_pe buy_zone.get(pe_max, np.inf) price_ok buy_zone.get(price_min, -np.inf) current_price buy_zone.get(price_max, np.inf) if pe_ok and price_ok: zone BUY_ZONE reason.append(fPE({current_pe:.1f})和价格({current_price})均进入理想买入区间) # 卖出警戒区判断 if sell_zone: if current_pe sell_zone.get(pe_min, np.inf) or current_price sell_zone.get(price_min, np.inf): zone SELL_ZONE reason.append(fPE({current_pe:.1f})或价格({current_price})触及高估警戒线) # 如果既不是买入区也不是卖出区就是HOLD区 if not reason: reason.append(处于持有观察区) return zone, ; .join(reason) def check_loss_alert(self, stock_code, daily_change): 检查日内亏损警报 if stock_code not in self.config: return False, threshold self.config[stock_code].get(loss_alert, -0.07) if daily_change threshold: # 跌幅达到或超过阈值 return True, f日内跌幅({daily_change:.2%})超过警报阈值({threshold:.0%})触发强制复查纪律 return False, def run_checks_for_stock(self, stock_code, market_data): 对单只股票执行所有纪律检查 # market_data 应包含实时行情、基本面数据等 results [] # 1. 仓位检查 (假设有计划买入金额) planned_buy 0 # 这里可以根据策略动态计算示例为0 pos_ok, pos_msg self.check_position_limit(stock_code, planned_buy) results.append((POSITION, pos_ok, pos_msg)) # 2. 估值检查 current_pe market_data.get(pe, 0) current_price market_data.get(price, 0) zone, zone_msg self.check_valuation_zone(stock_code, current_pe, current_price) results.append((VALUATION, zone, zone_msg)) # 3. 亏损警报检查 daily_change market_data.get(change, 0) loss_alert, loss_msg self.check_loss_alert(stock_code, daily_change) if loss_alert: results.append((LOSS_ALERT, False, loss_msg)) # 警报视为一种“不通过” else: results.append((LOSS_ALERT, True, 日内波动在正常范围内)) return results注意事项规则优先级在实际运行中需要定义规则的优先级。例如“仓位超限”是最高级别的禁止性规则一旦触发即使估值再诱人也不应买入。而“进入买入区”只是一个建议性规则还需要结合“趋势检查”等其他规则综合判断。参数动态调整估值区间buy_zone/sell_zone不是一成不变的。我每季度或每半年会根据公司最新财报和行业情况回顾并手动调整一次这体现了人的主观判断对系统的输入。布尔值与状态值规则检查的返回值设计很重要。像仓位、亏损警报这类“是/否”问题返回布尔值。像估值区间这种有多种状态的返回状态字符串如BUY_ZONE,SELL_ZONE,HOLD更清晰。3.3 通知与日志记录模块纪律必须被看见、被记录才能发挥作用。我使用钉钉机器人和本地文件日志。import requests import json from datetime import datetime class Notifier: def __init__(self, dingtalk_webhook_urlNone, log_filetrading_discipline.log): self.dingtalk_url dingtalk_webhook_url # 设置文件日志 self.file_logger logging.getLogger(file_logger) self.file_logger.setLevel(logging.INFO) file_handler logging.FileHandler(log_file, encodingutf-8) file_handler.setFormatter(logging.Formatter(%(asctime)s - %(levelname)s - %(message)s)) self.file_logger.addHandler(file_handler) def send_dingtalk_msg(self, title, text, is_at_allFalse): 发送钉钉机器人消息 if not self.dingtalk_url: return False headers {Content-Type: application/json} data { msgtype: markdown, markdown: { title: title, text: f**{title}**\n\n{text}\n\n 来自AI纪律小助手 {datetime.now().strftime(%H:%M:%S)} }, at: {isAtAll: is_at_all} } try: resp requests.post(self.dingtalk_url, headersheaders, datajson.dumps(data), timeout5) if resp.status_code 200: return True else: self.file_logger.error(f钉钉发送失败: {resp.status_code}, {resp.text}) return False except Exception as e: self.file_logger.error(f钉钉发送异常: {e}) return False def log_discipline_event(self, stock_code, event_type, result, detail): 记录纪律事件到文件和数据库如SQLite log_entry { timestamp: datetime.now(), stock_code: stock_code, event_type: event_type, # 如 ‘VALUATION_CHECK’, ‘LOSS_ALERT’ result: str(result), # 如 ‘BUY_ZONE’, ‘False’ detail: detail } # 写入日志文件 self.file_logger.info(f{stock_code} - {event_type} - {result} - {detail}) # 这里可以同时写入SQLite数据库便于后续按股票、按时间查询分析 # self._write_to_db(log_entry) # 根据事件重要性决定是否发送即时通知 if event_type LOSS_ALERT and result is False: self.send_dingtalk_msg(f⚠️ 纪律警报{stock_code}, detail) elif event_type VALUATION_CHECK and result SELL_ZONE: self.send_dingtalk_msg(f 估值提醒{stock_code}, detail) # 示例SQLite日志存储可选但推荐 import sqlite3 class DisciplineLogger: def __init__(self, db_pathdiscipline_log.db): self.conn sqlite3.connect(db_path) self._init_db() def _init_db(self): cursor self.conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS log ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME, stock_code TEXT, event_type TEXT, result TEXT, detail TEXT ) ) self.conn.commit() def write_log(self, log_entry): cursor self.conn.cursor() cursor.execute( INSERT INTO log (timestamp, stock_code, event_type, result, detail) VALUES (?, ?, ?, ?, ?) , (log_entry[timestamp], log_entry[stock_code], log_entry[event_type], log_entry[result], log_entry[detail])) self.conn.commit()实操心得分级通知不要所有信息都推送手机否则很快就会被无视。我只对重要警报如亏损超阈值、触及高估线和关键状态变化如从HOLD区进入BUY_ZONE进行钉钉推送。日常的、正常的检查结果只记录到本地日志供复盘时查看。结构化日志将日志写入SQLite数据库而非纯文本文件在后期进行数据分析时优势巨大。你可以轻松地写出SQL查询比如“统计过去一个月内某只股票触发‘禁止买入’纪律的次数”从而量化评估自己的冲动倾向和纪律的有效性。3.4 主循环与调度模块最后我们需要一个“指挥官”把以上模块串联起来并让它们定时运行。import schedule import time class AIDisciplineAssistant: def __init__(self, config_pathconfig.json): self.load_config(config_path) self.fetcher DataFetcher() self.engine DisciplineEngine(self.config) self.notifier Notifier(dingtalk_webhook_urlYOUR_WEBHOOK_URL) self.logger DisciplineLogger() def load_config(self, path): # 从JSON文件加载配置 import json with open(path, r, encodingutf-8) as f: self.config json.load(f) def monitor_single_stock(self, stock_code): 监控单只股票的完整流程 # 1. 获取数据 quote self.fetcher.get_realtime_quote(stock_code) if not quote: return basic_info self.fetcher.get_basic_info(stock_code) or {} # 2. 组装市场数据 market_data { price: quote[latest], change: quote[change] / 100, # 转换为小数 pe: basic_info.get(pe_ttm, 0), pb: basic_info.get(pb, 0) } # 3. 执行纪律检查 check_results self.engine.run_checks_for_stock(stock_code, market_data) # 4. 记录和通知 for check_type, result, detail in check_results: log_entry { timestamp: datetime.now(), stock_code: stock_code, event_type: check_type, result: str(result), detail: f{detail} | 股价:{quote[latest]} } self.logger.write_log(log_entry) # 根据结果类型和严重程度决定是否推送 if check_type LOSS_ALERT and result is False: self.notifier.send_dingtalk_msg(f亏损警报 {stock_code}, detail) elif check_type VALUATION_CHECK and result SELL_ZONE: self.notifier.send_dingtalk_msg(f高估警戒 {stock_code}, detail) def run_daily_monitoring(self): 每日收盘后执行一次用于盘后复盘和数据更新 logger.info(开始执行盘后纪律检查...) for stock_code in self.config.keys(): self.monitor_single_stock(stock_code) time.sleep(1) # 短暂停顿避免请求过快 logger.info(盘后纪律检查完成。) def run_intraday_monitoring(self): 盘中定时监控频率可以高一些如每30分钟 logger.info(开始执行盘中纪律检查...) for stock_code in self.config.keys(): self.monitor_single_stock(stock_code) time.sleep(1) logger.info(盘中纪律检查完成。) def start(self): 启动定时任务 # 每个交易日开盘前9:15运行一次 schedule.every().day.at(09:15).do(self.run_daily_monitoring).tag(daily_pre_open) # 盘中每30分钟运行一次 schedule.every(30).minutes.do(self.run_intraday_monitoring).tag(intraday) # 每个交易日收盘后15:30运行一次复盘 schedule.every().day.at(15:30).do(self.run_daily_monitoring).tag(daily_post_close) logger.info(AI纪律小助手已启动...) while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次是否有任务需要执行 if __name__ __main__: assistant AIDisciplineAssistant(my_stock_config.json) # 测试单次运行 # assistant.monitor_single_stock(sh.600519) # 正式运行启动定时任务 assistant.start()关键点解析定时策略使用schedule库实现定时任务非常轻便。我设置了三个主要任务开盘前预演、盘中监控、收盘后复盘。盘中监控频率不宜过高30分钟到1小时一次足以捕捉关键的价格异动同时避免给数据源造成压力。配置外置所有股票的自定义规则估值区间、仓位上限等都放在外部的config.json文件中。这样修改规则时无需改动代码也更安全。循环与休眠主循环while True配合time.sleep(60)让程序持续运行。在监控大量股票时循环内要加入适当的sleep体现良好的网络礼仪。4. 常见问题、踩坑记录与优化方向在实际搭建和运行这个小助手的过程中我遇到了不少问题也总结了一些经验。4.1 数据源稳定性与替代方案问题最初我尝试使用yfinance但网络延迟和稳定性时常出问题特别是对于A股数据支持不佳。akshare虽然强大但部分接口也可能因网站改版而暂时失效。解决方案多源备份对于核心的实时行情我准备了两个数据源函数。当主源akshare.stock_zh_a_spot_em失败时自动尝试备用源如baostock或通过requests直接访问某些财经网站的API。重试机制在数据获取函数中加入简单的重试逻辑。def fetch_with_retry(func, *args, retries3, delay2, **kwargs): for i in range(retries): try: return func(*args, **kwargs) except Exception as e: logger.warning(f第{i1}次尝试失败: {e}) if i retries - 1: time.sleep(delay) logger.error(f所有{retries}次尝试均失败) return None数据缓存与离线将每日获取的历史K线数据存储到本地SQLite数据库。这样即使当天网络完全中断系统依然能基于昨日收盘价和历史数据进行大部分纪律检查估值、仓位等只是实时性稍差。4.2 规则冲突与优先级处理问题当多条规则对同一只股票给出相反的建议时怎么办比如股价跌入了“理想买入区”触发买入建议但同时该股票日内跌幅已超过-7%触发亏损警报要求强制复查暂停一切操作。我的处理逻辑 我制定了一个简单的规则优先级矩阵在DisciplineEngine中增加一个仲裁函数规则类型优先级输出建议说明仓位超限最高 (1)禁止买入/加仓风控底线无条件执行亏损警报高 (2)暂停并强制复查暂停所有计划操作检查原因高估警戒中 (3)建议减持/不买入提醒风险但不强制进入买入区低 (4)建议关注/可买入需结合其他条件如趋势在代码中检查按优先级顺序进行。一旦高优先级规则触发返回False或警报状态则立即终止后续低优先级规则的“建议性输出”并以高优先级规则的结果为准。例如只要“亏损警报”触发无论是否在“买入区”最终输出都是“暂停操作强制复查”。4.3 避免“过度自动化”与保持人的最终决策权这是一个纪律辅助系统而非全自动交易系统。我坚决避免让它直接连接券商API进行实盘交易。原因有三复杂性实盘交易涉及资金安全、订单类型、成交处理、异常恢复等极其复杂的问题个人项目很难做到工业级可靠。灵活性市场有很多规则无法量化的情形如突发的政策变化、公司管理层丑闻等需要人的主观判断。责任归属代码出错导致亏损责任难以界定。因此本系统的所有输出都是“建议”和“警报”。最终的买卖按钮必须由我亲手点击。系统的价值在于在我点击按钮前强制我阅读并确认这些纪律检查结果像一道必须经过的安检门。4.4 性能优化与日志分析当监控股票数量增多比如超过50只且历史数据拉取周期很长时程序可能会变慢。优化措施异步IO使用asyncio和aiohttp并发获取多只股票的实时数据可以极大缩短数据获取时间。增量更新历史数据每天只收盘后全量更新一次。盘中计算指标如20日均线时只需在本地历史数据基础上追加最新的一条行情进行计算无需重新拉取全部历史。定期复盘日志每周或每月我会用pandas读取SQLite中的日志表做一些简单的分析import pandas as pd import sqlite3 conn sqlite3.connect(discipline_log.db) df pd.read_sql_query(SELECT * FROM log WHERE event_typeLOSS_ALERT, conn) # 统计哪只股票触发亏损警报最频繁 alert_count df.groupby(stock_code).size().sort_values(ascendingFalse) print(亏损警报触发统计) print(alert_count)这能帮助我发现哪些股票波动性过大是否不适合我的投资体系或者我的警报阈值设置是否不合理。4.5 情绪分析与信息过滤的初步尝试除了价格和财务数据市场情绪也是影响决策的重要因素。我尝试了一个非常简单的情绪分析模块作为补充from snownlp import SnowNLP class SimpleSentimentAnalyzer: def __init__(self): self.alert_keywords [暴涨, 崩盘, 史诗级, 重大利好, 突发, 危险信号] def analyze_news_title(self, title): 分析新闻标题情绪 # 方法1关键词匹配 for kw in self.alert_keywords: if kw in title: return HIGH_EMOTION, f标题包含情绪化关键词 {kw} # 方法2简单情感分析基于SnowNLP对财经新闻效果有限仅供参考 try: s SnowNLP(title) sentiment s.sentiments # 0-1, 越接近1越正面 if sentiment 0.7: return POSITIVE, f标题情感偏积极({sentiment:.2f}) elif sentiment 0.3: return NEGATIVE, f标题情感偏消极({sentiment:.2f}) else: return NEUTRAL, f标题情感中性({sentiment:.2f}) except: return UNKNOWN, 情感分析失败 def filter_and_alert(self, news_list): 过滤新闻列表并发送警报 for news in news_list: sentiment, reason self.analyze_news_title(news[title]) if sentiment in [HIGH_EMOTION, NEGATIVE]: # 记录日志或发送通知 self.notifier.log_event(NEWS_ALERT, news[stock_code], False, f{reason}: {news[title]})注意情绪分析在投资中是一把双刃剑准确性很难保证。我仅将其作为一个“噪音探测器”当标题出现极度情绪化词汇时提醒我“注意你正在阅读一条可能煽动情绪的消息请冷静。” 绝不将其作为买卖依据。5. 项目总结与个人体会搭建并运行这个“AI股票小助手”大半年以来它没有帮我抓住任何一个涨停板但它确实帮我管住了好几次想要“手痒”操作的手。最典型的一次是某只重仓股在利好消息下快速冲高一度触及我设定的“高估警戒线”钉钉机器人立刻发来消息。当时我内心非常想“再等等也许还能更高”但屏幕上那条冰冷的纪律提示不断提醒我之前的计划。最终我选择部分减仓后来股价回落那次纪律执行让我保住了相当一部分利润。这个过程让我深刻体会到纪律的本质是对抗本能投资中最大的成本往往是情绪成本。这个系统的作用就是在情绪升温时给你泼一盆代码编写的“冷水”让你回到事先规划好的路线上。代码是投资体系的镜子当你试图将投资逻辑写成代码时会发现很多自己原来以为想清楚的事情其实模糊不清。比如“低估”到底怎么定义是PE低于20倍还是PB低于行业均值这个过程逼着你把投资体系细化、量化、逻辑化本身就是一次巨大的提升。它是一个动态进化的工具我的配置config.json已经修改了十几个版本。每次市场风格切换或者自己读了新的投资理念都会促使我回头调整那些规则参数。这个系统也在随着我的认知一起成长。降低关注频率提升决策质量以前恨不得每分钟都看盘现在有了这个助手在后台默默监控我只需要在它“尖叫”发出警报时去看一眼。这让我从市场的短期噪音中解放出来能把更多时间花在研究公司基本面和行业长期趋势上。最后想说的是这个项目没有任何神秘的地方用到的都是最基础的Python知识。它的威力不在于算法有多高级而在于它将一种正确的投资行为模式固化了。对于像我一样在市场中不断与自身人性弱点斗争的普通投资者来说这不失为一种笨拙但有效的“科技赋能”。如果你也受困于纪律问题不妨从定义一个最简单的规则开始比如“任何单日跌幅超过5%的股票当天不再加仓”然后试着用代码把它实现出来。这个过程或许比你从代码中学到的更多。