1717import uuid
1818from collections import defaultdict
1919from concurrent .futures import ThreadPoolExecutor , as_completed
20- from datetime import date , timedelta
20+ from datetime import date , datetime , timedelta , timezone
2121from typing import List , Dict , Any , Optional , Tuple
2222
2323import pandas as pd
3939from src .services .social_sentiment_service import SocialSentimentService
4040from src .enums import ReportType
4141from src .stock_analyzer import StockTrendAnalyzer , TrendAnalysisResult
42- from src .core .trading_calendar import get_market_for_stock , is_market_open
42+ from src .core .trading_calendar import (
43+ get_effective_trading_date ,
44+ get_market_for_stock ,
45+ get_market_now ,
46+ is_market_open ,
47+ )
4348from data_provider .us_index_mapping import is_us_stock_code
4449from bot .models import BotMessage
4550
@@ -135,19 +140,21 @@ def __init__(
135140 def fetch_and_save_stock_data (
136141 self ,
137142 code : str ,
138- force_refresh : bool = False
143+ force_refresh : bool = False ,
144+ current_time : Optional [datetime ] = None ,
139145 ) -> Tuple [bool , Optional [str ]]:
140146 """
141147 获取并保存单只股票数据
142148
143149 断点续传逻辑：
144- 1. 检查数据库是否已有今日数据
150+ 1. 检查数据库是否已有最新可复用交易日数据
145151 2. 如果有且不强制刷新，则跳过网络请求
146152 3. 否则从数据源获取并保存
147153
148154 Args:
149155 code: 股票代码
150156 force_refresh: 是否强制刷新（忽略本地缓存）
157+ current_time: 本轮运行冻结的参考时间，用于统一断点续传目标交易日判断
151158
152159 Returns:
153160 Tuple[是否成功, 错误信息]
@@ -157,16 +164,15 @@ def fetch_and_save_stock_data(
157164 # 首先获取股票名称
158165 stock_name = self .fetcher_manager .get_stock_name (code , allow_realtime = False )
159166
160- today = date .today ()
161- # 注意：这里用自然日 date.today() 做“断点续传”判断。
162- # 若在周末/节假日/非交易日运行，或机器时区不在中国，可能出现：
163- # - 数据库已有最新交易日数据但仍会重复拉取（has_today_data 返回 False）
164- # - 或在跨日/时区偏移时误判“今日已有数据”
165- # 该行为目前保留（按需求不改逻辑），但如需更严谨可改为“最新交易日/数据源最新日期”判断。
166-
167- # 断点续传检查：如果今日数据已存在，跳过
168- if not force_refresh and self .db .has_today_data (code , today ):
169- logger .info (f"{ stock_name } { code } )
167+ target_date = self ._resolve_resume_target_date (
168+ code , current_time = current_time
169+ )
170+
171+ # 断点续传检查：如果最新可复用交易日的数据已存在，则跳过
172+ if not force_refresh and self .db .has_today_data (code , target_date ):
173+ logger .info (
174+ f"{ stock_name } { code } { target_date }
175+ )
170176 return True , None
171177
172178 # 从数据源获取数据
@@ -295,7 +301,8 @@ def analyze_stock(self, code: str, report_type: ReportType, query_id: str) -> Op
295301 # Step 3: 趋势分析（基于交易理念）— 在 Agent 分支之前执行，供两条路径共用
296302 trend_result : Optional [TrendAnalysisResult ] = None
297303 try :
298- end_date = date .today ()
304+ _mkt = get_market_for_stock (normalize_stock_code (code ))
305+ end_date = get_market_now (_mkt ).date ()
299306 start_date = end_date - timedelta (days = 89 ) # ~60 trading days for MA60
300307 historical_bars = self .db .get_data_range (code , start_date , end_date )
301308 if historical_bars :
@@ -379,10 +386,13 @@ def analyze_stock(self, code: str, report_type: ReportType, query_id: str) -> Op
379386
380387 if context is None :
381388 logger .warning (f"{ stock_name } { code } )
389+ _mkt_date = get_market_now (
390+ get_market_for_stock (normalize_stock_code (code ))
391+ ).date ()
382392 context = {
383393 'code' : code ,
384394 'stock_name' : stock_name ,
385- 'date' : date . today () .isoformat (),
395+ 'date' : _mkt_date .isoformat (),
386396 'data_missing' : True ,
387397 'today' : {},
388398 'yesterday' : {}
@@ -566,7 +576,9 @@ def _enhance_context(
566576 enhanced ['ma_status' ] = self ._compute_ma_status (
567577 price , trend_result .ma5 , trend_result .ma10 , trend_result .ma20
568578 )
569- enhanced ['date' ] = date .today ().isoformat ()
579+ enhanced ['date' ] = get_market_now (
580+ get_market_for_stock (normalize_stock_code (enhanced .get ('code' , '' )))
581+ ).date ().isoformat ()
570582 if yesterday_close is not None :
571583 try :
572584 yc = float (yesterday_close )
@@ -950,7 +962,8 @@ def _augment_historical_with_realtime(
950962 if not enable_realtime_tech :
951963 return df
952964 market = get_market_for_stock (code )
953- if market and not is_market_open (market , date .today ()):
965+ market_today = get_market_now (market ).date ()
966+ if market and not is_market_open (market , market_today ):
954967 return df
955968
956969 last_val = df ['date' ].max ()
@@ -968,7 +981,7 @@ def _augment_historical_with_realtime(
968981 amt = getattr (realtime_quote , 'amount' , None )
969982 pct = getattr (realtime_quote , 'change_pct' , None )
970983
971- if last_date >= date . today () :
984+ if last_date >= market_today :
972985 # Update last row with realtime close (copy to avoid mutating caller's df)
973986 df = df .copy ()
974987 idx = df .index [- 1 ]
@@ -989,7 +1002,7 @@ def _augment_historical_with_realtime(
9891002 # Append virtual today row
9901003 new_row = {
9911004 'code' : code ,
992- 'date' : date . today () ,
1005+ 'date' : market_today ,
9931006 'open' : open_p ,
9941007 'high' : high_p ,
9951008 'low' : low_p ,
@@ -1019,6 +1032,16 @@ def _build_context_snapshot(
10191032 "chip_distribution_raw" : self ._safe_to_dict (chip_data ),
10201033 }
10211034
1035+ @staticmethod
1036+ def _resolve_resume_target_date (
1037+ code : str , current_time : Optional [datetime ] = None
1038+ ) -> date :
1039+ """
1040+ Resolve the trading date used by checkpoint/resume checks.
1041+ """
1042+ market = get_market_for_stock (normalize_stock_code (code ))
1043+ return get_effective_trading_date (market , current_time = current_time )
1044+
10221045 @staticmethod
10231046 def _safe_to_dict (value : Any ) -> Optional [Dict [str , Any ]]:
10241047 """
@@ -1092,6 +1115,7 @@ def process_single_stock(
10921115 single_stock_notify : bool = False ,
10931116 report_type : ReportType = ReportType .SIMPLE ,
10941117 analysis_query_id : Optional [str ] = None ,
1118+ current_time : Optional [datetime ] = None ,
10951119 ) -> Optional [AnalysisResult ]:
10961120 """
10971121 处理单只股票的完整流程
@@ -1110,6 +1134,7 @@ def process_single_stock(
11101134 skip_analysis: 是否跳过 AI 分析
11111135 single_stock_notify: 是否启用单股推送模式（每分析完一只立即推送）
11121136 report_type: 报告类型枚举（从配置读取，Issue #119）
1137+ current_time: 本轮运行冻结的参考时间，用于统一断点续传目标交易日判断
11131138
11141139 Returns:
11151140 AnalysisResult 或 None
@@ -1118,7 +1143,9 @@ def process_single_stock(
11181143
11191144 try :
11201145 # Step 1: 获取并保存数据
1121- success , error = self .fetch_and_save_stock_data (code )
1146+ success , error = self .fetch_and_save_stock_data (
1147+ code , current_time = current_time
1148+ )
11221149
11231150 if not success :
11241151 logger .warning (f"[{ code } { error } )
@@ -1197,6 +1224,9 @@ def run(
11971224 logger .info (f"===== 开始分析 { len (stock_codes )} )
11981225 logger .info (f"股票列表: { ', ' .join (stock_codes )} )
11991226 logger .info (f"并发数: { self .max_workers } { '仅获取数据' if dry_run else '完整分析' } )
1227+
1228+ # 冻结本轮运行的统一参考时间，避免跨市场收盘边界时同批股票使用不同目标交易日。
1229+ resume_reference_time = datetime .now (timezone .utc )
12001230
12011231 # === 批量预取实时行情（优化：避免每只股票都触发全量拉取）===
12021232 # 只有股票数量 >= 5 时才进行预取，少量股票直接逐个查询更高效
@@ -1243,6 +1273,7 @@ def run(
12431273 single_stock_notify = False ,
12441274 report_type = report_type , # Issue #119: 传递报告类型
12451275 analysis_query_id = uuid .uuid4 ().hex ,
1276+ current_time = resume_reference_time ,
12461277 ): code
12471278 for code in stock_codes
12481279 }
@@ -1278,8 +1309,17 @@ def run(
12781309
12791310 # dry-run 模式下，数据获取成功即视为成功
12801311 if dry_run :
1281- # 检查哪些股票的数据今天已存在
1282- success_count = sum (1 for code in stock_codes if self .db .has_today_data (code ))
1312+ # 检查哪些股票的最新可复用交易日数据已存在
1313+ success_count = sum (
1314+ 1
1315+ for code in stock_codes
1316+ if self .db .has_today_data (
1317+ code ,
1318+ self ._resolve_resume_target_date (
1319+ code , current_time = resume_reference_time
1320+ ),
1321+ )
1322+ )
12831323 fail_count = len (stock_codes ) - success_count
12841324 else :
12851325 success_count = len (results )
0 commit comments