import pandas as pd import numpy as np from datetime import datetime from datetime import timedelta import pymysql from sqlalchemy import create_engine from flask import Flask, request, jsonify from dateutil.relativedelta import relativedelta import logging from multiprocessing import Pool, cpu_count import time import os total_start = time.time() # 在文件开头添加日志配置 def setup_logging(): log_dir = os.path.join(os.path.dirname(__file__), 'logs') os.makedirs(log_dir, exist_ok=True) logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(os.path.join(log_dir, 'api.log')), logging.StreamHandler() ] ) setup_logging() logger = logging.getLogger(__name__) # 在API.py开头添加 import sys # 新增函数：处理单个日期的卦象计算 def calculate_day_gua(args): date, Map64Gua, Map24Jieqi = args try: _, Gua1Day2, _, _, _ = guaCalc_huangjijingshi(Map64Gua, Map24Jieqi, date) return { 'date': date, 'Year': date.year, 'Trigram': Gua1Day2.trigram if hasattr(Gua1Day2, 'trigram') else None } except Exception as e: logger.error(f"计算日期 {date} 的卦象时出错: {str(e)}") return { 'date': date, 'Year': date.year, 'Trigram': None } # 修改日卦计算部分 def calculate_daily_gua(Map64Gua, Map24Jieqi, date_range): # 准备参数 args = [(date, Map64Gua, Map24Jieqi) for date in date_range] # 使用进程池并行计算 with Pool(processes=min(6, cpu_count())) as pool: results = pool.map(calculate_day_gua, args) return pd.DataFrame(results) # 替换原来的get_user_date函数 def get_user_date(): if len(sys.argv) > 1: datetime_input = sys.argv[1] try: return datetime.strptime(datetime_input, '%Y-%m-%d %H:%M:%S') except ValueError: print("日期格式不正确，请使用YYYY-MM-DD HH:MM:SS格式", file=sys.stderr) sys.exit(1) else: print("没有传入日期时间参数", file=sys.stderr) sys.exit(1) # 导入自定义卦象计算模块 from guaCalc_huangjijingshi import guaCalc_huangjijingshi from luckCalc_huangjijingshi import luckCalc_huangjijingshi # ========== 数据库连接配置 ========== # 你的账号、密码、主机、端口 username = 'cn_ainvest_db' password = 'cn_ainvest_sd3a1' host = 'rm-2zewagytttzk6f24xno.mysql.rds.aliyuncs.com' port = 3306 database = 'ai_strategy' # 这里改成你要的数据库 # 创建 SQLAlchemy engine lianjie_start = time.time() engine = create_engine(f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}') logger.info(f"数据框连接完成，耗时: {time.time() - lianjie_start:.2f}秒") # ========== 文件路径配置 ========== configPath = r'C:\AI trading\config\Rey\test_reinforcement' matlabPath = r'D:\Dropbox\Matlab\Rey\MATLAB' # 修改为 Excel 文件路径 tempPath = r'C:\Users\24011\Documents\WeChat Files\wxid_k4ep58f81rx421\FileStorage\File\2025-04\tuigua_huangjijingshi - 副本\tuigua_huangjijingshi - 副本\皇极经世.xlsx' # ========== 加载 64 卦映射表 ========== Map64Gua = pd.read_excel(tempPath, sheet_name="bagua") # 转换爻数据为整数 for col in ['yao1', 'yao2', 'yao3', 'yao4', 'yao5', 'yao6']: Map64Gua[col] = Map64Gua[col].apply(lambda x: int(x) if str(x).isdigit() else x) Map64GuaOmit = Map64Gua[Map64Gua['change_omit'] == 0] # ========== 从数据库加载 24 节气数据 ========== def load_solar_terms(conn_params): """通过 SQLAlchemy 连接读取 solar_terms 表""" sqlquery = "SELECT * FROM solar_terms WHERE As_Of_Date > '1744-01-01'" df = pd.read_sql(sqlquery, engine) # 尝试将 As_Of_Date 转为 datetime，如果失败就变成 NaT df['As_Of_Date'] = pd.to_datetime(df['As_Of_Date'], errors='coerce') # 过滤掉早于 1900-01-01 或转换失败的日期 df = df[df['As_Of_Date'] >= pd.Timestamp('1780-01-01')] df = df.dropna(subset=['As_Of_Date']) # df['As_Of_Date'] = df['As_Of_Date'] df = df.sort_values('As_Of_Date') return df # ========== 用户输入日期 ========== # def get_user_date(): # if len(sys.argv) > 1: # 命令行参数方式 # date_input = sys.argv[1] # else: # 标准输入方式 # date_input = sys.stdin.read().strip() # try: # year, month, day = map(int, date_input.split('-')) # return datetime(year, month, day, 0, 0, 0) # except ValueError: # print("日期格式不正确，请使用YYYY-MM-DD格式", file=sys.stderr) # sys.exit(1) # 获取用户输入 def get_user_date(): # date_input = sys.stdin.read().strip() if len(sys.argv) == 1 else sys.argv[1] if len(sys.argv) > 1: datetime_input = sys.argv[1] else: print("没有传入日期时间参数", file=sys.stderr) sys.exit(1) try: # 解析包含时间的日期 return datetime.strptime(datetime_input, '%Y-%m-%d %H:%M:%S') # # 确保日期格式为 YYYY-MM-DD # return datetime.strptime(date_input, '%Y-%m-%d') except ValueError: print("日期格式不正确，请使用YYYY-MM-DD格式", file=sys.stderr) sys.exit(1) # 加载数据并标记重要节气 load_start = time.time() solar_terms = load_solar_terms(engine) logger.info(f"节气加载完成，耗时: {time.time() - load_start:.2f}秒") important_terms = ['冬至', '雨水', '谷雨', '夏至', '处暑', '霜降'] solar_terms['isImportant'] = solar_terms['Solar_Terms'].isin(important_terms).astype(int) Map24Jieqi = solar_terms.copy() # ========== 读取CSV文件并处理每个日期的卦象 ========== read_start = time.time() csv_path = r'C:\Users\24011\Documents\WeChat Files\wxid_k4ep58f81rx421\FileStorage\File\2025-04\tuigua_huangjijingshi - 副本\tuigua_huangjijingshi - 副本\python1\2020 - 2030年每天卦.csv' table = pd.read_csv(csv_path) logger.info(f"读取csv文件完成，耗时: {time.time() - read_start:.2f}秒") try: piwen_path = r'C:\Users\24011\Documents\WeChat Files\wxid_k4ep58f81rx421\FileStorage\File\2025-04\tuigua_huangjijingshi - 副本\tuigua_huangjijingshi - 副本\python1\批文.xlsx' piwen_data = pd.read_excel(piwen_path) # 打印列名用于调试 print("批文文件列名:", piwen_data.columns.tolist(), file=sys.stderr) # 检查必要的列是否存在 if not all(col in piwen_data.columns for col in ['卦名', '命卦批文', '八个字基本盘']): print("批文文件缺少必要列，请检查Excel文件列名", file=sys.stderr) piwen_data = pd.DataFrame(columns=['卦名', '命卦批文', '八个字基本盘']) except Exception as e: print(f"读取批文文件失败: {str(e)}", file=sys.stderr) piwen_data = pd.DataFrame(columns=['卦名', '命卦批文', '八个字基本盘']) def get_piwen_info(trigram_name): """根据卦名获取批文和基本盘信息""" if not isinstance(trigram_name, str): trigram_name = str(trigram_name) try: match = piwen_data[piwen_data['卦名'] == trigram_name] if not match.empty: return { 'piwen': match.iloc[0]['命卦批文'], 'basic_info': match.iloc[0]['八个字基本盘'] } else: print(f"未找到卦名 {trigram_name} 的批文信息", file=sys.stderr) return {'piwen': '暂无批文信息', 'basic_info': '暂无基本盘信息'} except Exception as e: print(f"获取批文信息出错: {str(e)}", file=sys.stderr) return {'piwen': '暂无批文信息', 'basic_info': '暂无基本盘信息'} # 获取用户输入 kDate = get_user_date() #print("\n计算日期:", kDate.strftime('%Y-%m-%d')) # 计算卦象 Gua1Hour, Gua1Day, Gua1Month, Gua1Year, GuaLuck = guaCalc_huangjijingshi( Map64Gua, Map24Jieqi, kDate ) # print("日卦:", Gua1Day) import json # 导入 JSON 库 # ========== 计算并返回 JSON 格式的结果 ========== # ========== 计算并返回 JSON 格式的结果 ========== if __name__ == "__main__": print("=== 调试开始 ===", file=sys.stderr) # 打印到 stderr 不会干扰 stdout 的 JSON try: input_start = time.time() kDate = get_user_date() logger.info(f"获取用户输入完成，耗时: {time.time() - input_start:.2f}秒") # 计算卦象 calc_start = time.time() Gua1Hour, Gua1Day, Gua1Month, Gua1Year, GuaLuck = guaCalc_huangjijingshi( Map64Gua, Map24Jieqi, kDate ) # 添加详细的年卦日志 logger.info("\n【当前日期年卦详细计算】") logger.info(f"计算日期: {kDate}") logger.info(f"年卦类型: {type(Gua1Year)}") if isinstance(Gua1Year, dict): logger.info(f"卦名字典内容: {Gua1Year}") logger.info(f"卦名: {Gua1Year.get('trigram', '未找到')}") logger.info(f"爻位: {Gua1Year.get('yao1', '')}{Gua1Year.get('yao2', '')}{Gua1Year.get('yao3', '')}" f"{Gua1Year.get('yao4', '')}{Gua1Year.get('yao5', '')}{Gua1Year.get('yao6', '')}") logger.info(f"二进制值: {Gua1Year.get('value_2binary', '')}") elif hasattr(Gua1Year, 'trigram'): logger.info(f"卦名(对象属性): {Gua1Year.trigram}") logger.info(f"爻位: {Gua1Year.yao1}{Gua1Year.yao2}{Gua1Year.yao3}{Gua1Year.yao4}{Gua1Year.yao5}{Gua1Year.yao6}") logger.info(f"二进制值: {Gua1Year.value_2binary}") else: logger.error("无法识别的年卦数据类型") logger.info(f"卦象计算完成，耗时: {time.time() - calc_start:.2f}秒") logger.info("月卦:", Gua1Month) logger.info("日卦:", Gua1Day) # print(Gua1Day.apply(type)) logger.info("时卦:", Gua1Hour) logger.info("吉凶卦:", GuaLuck) logger.info(f"卦象计算完成，耗时: {time.time() - calc_start:.2f}秒") # # 计算10年前（当前日期减10年） # Yearpre10 = kDate.replace(year=kDate.year - 10) # # 计算10年后（当前日期加10年） # Yearpast10 = kDate.replace(year=kDate.year + 10) # 或者使用relativedelta（更精确处理闰年等情况） # 计算年份卦象 year_start = time.time() Yearpre10 = datetime(2010, 1, 1) Yearpast10 = datetime(2030, 1, 1,) # 生成日期范围（修正end参数） date_range = pd.date_range( start=Yearpre10.replace(month=1, day=1), # 确保从1月1日开始 end=Yearpast10.replace(month=1, day=1), # 确保到1月1日结束 freq='YS' # 每年第一天 ) #date_range = pd.date_range(start=Yearpre10, end=Yearpast10, freq='YS') # 每年第一天 year_gua_list = [] for date in date_range: _, _, _, Gua1Year_2, _ = guaCalc_huangjijingshi(Map64Gua, Map24Jieqi, date) year_gua_list.append({ 'date': date, 'Year': date.year, 'Trigram': Gua1Year_2.trigram if hasattr(Gua1Year_2, 'trigram') else None, }) logger.info(f"用于计算年卦的日期参数: {kDate}") # 确保与前端一致 yearGuaMap = pd.DataFrame(year_gua_list) logger.info("yearGuaMap:", yearGuaMap) #========== 计算年份吉凶 ========== # 计算年份吉凶 luck_start = time.time() LuckYear = luckCalc_huangjijingshi(Map64Gua, yearGuaMap, GuaLuck) #月卦 todayyear = datetime.today().year Yearpre1 = todayyear - 1 Yearpast1 = todayyear + 1 date_range2 = pd.date_range( start=f'{Yearpre1}-01-01', # 设置开始时间为前一年1月1日 end=f'{Yearpast1}-12-01', # 设置结束时间为后一年12月1日 freq='MS' # 每月的第一天 ) year_gua_list_month = [] for date in date_range2: _, _, Gua1Year_3, _, _ = guaCalc_huangjijingshi(Map64Gua, Map24Jieqi, date) # 调试信息 logger.info(f"Gua1Year_3 type: {type(Gua1Year_3)}") if isinstance(Gua1Year_3, pd.DataFrame): logger.info(f"DataFrame columns: {Gua1Year_3.columns}") logger.info(f"First row: {Gua1Year_3.iloc[0].to_dict()}") month_label = date.strftime('%Y-%m') # 安全获取 trigram 值 trigram_value = None if isinstance(Gua1Year_3, pd.DataFrame) and not Gua1Year_3.empty: trigram_value = Gua1Year_3.iloc[0]['trigram'] if 'trigram' in Gua1Year_3.columns else None elif hasattr(Gua1Year_3, 'trigram'): trigram_value = Gua1Year_3.trigram year_gua_list_month.append({ 'month': month_label, 'Year': date.year, 'Trigram': trigram_value, }) yearGuaMap2 = pd.DataFrame(year_gua_list_month) # 计算月份吉凶 LuckYear_month = luckCalc_huangjijingshi(Map64Gua, yearGuaMap2, GuaLuck) # 添加月份列 (格式为"YYYY-MM") if isinstance(LuckYear_month, pd.DataFrame) and 'Year' in LuckYear_month.columns: # 创建月份列表 (1-12) months = [f"{year}-{str(month).zfill(2)}" for year in LuckYear_month['Year'].unique() for month in range(1, 13)] # 确保月份数量匹配 if len(months) >= len(LuckYear_month): LuckYear_month['month'] = months[:len(LuckYear_month)] else: # 如果月份不足，循环使用前几个月份 LuckYear_month['month'] = [months[i % len(months)] for i in range(len(LuckYear_month))] logger.info("添加月份列后的LuckYear_month:") logger.info(LuckYear_month.head()) else: logger.warning("LuckYear_month不是DataFrame或缺少Year列，无法添加月份信息") #日卦 # todayyear = datetime.today().year # date_range3 = pd.date_range( # start=f'{todayyear}-01-01', # 设置开始时间为今年1月1日 # end=f'{todayyear}-12-31', # 设置结束时间为今年12月31日 # freq='D' # 每天 # ) # year_gua_list_day = [] # # for date in date_range3: # # _, Gua1Day2, _, _, _ = guaCalc_huangjijingshi(Map64Gua, Map24Jieqi, date) # # year_gua_list_day.append({ # # 'date': date, # # 'Year': date.year, # # 'Trigram': Gua1Day2.trigram if hasattr(Gua1Day2, 'trigram') else None, # # }) # year_gua_list_day = calculate_daily_gua(Map64Gua, Map24Jieqi, date_range3) # yearGuaMap3 = pd.DataFrame(year_gua_list_day) # # 计算每日吉凶 # LuckYear_day = luckCalc_huangjijingshi(Map64Gua, yearGuaMap3, GuaLuck) # logger.info("添加日期前的LuckYear_day:") # logger.info(LuckYear_day.head()) # if isinstance(LuckYear_day, pd.DataFrame): # # 确保索引一致 # LuckYear_day = LuckYear_day.set_index(yearGuaMap3.index) # LuckYear_day['date'] = yearGuaMap3['date'] # logger.info("添加日期列后的LuckYear_day:") # logger.info(LuckYear_day.head()) # logger.info(f"吉凶计算完成，耗时: {time.time() - luck_start:.2f}秒") # # 检查 Gua1Day 的类型并正确处理 # if isinstance(Gua1Day, pd.DataFrame): # # 如果是 DataFrame，提取第一行 # day_data = Gua1Day.iloc[0] # elif isinstance(Gua1Day, pd.Series): # # 如果是 Series，直接使用 # day_data = Gua1Day # elif isinstance(Gua1Day, dict): # # 如果是字典，直接使用 # day_data = Gua1Day # else: # raise ValueError("Gua1Day 的类型不支持，必须是 DataFrame、Series 或 dict") def format_gua_data(gua_data): """通用格式化卦象数据的函数""" if isinstance(gua_data, pd.DataFrame): # DataFrame类型 - 取第一行转为字典 data = gua_data.iloc[0].to_dict() elif isinstance(gua_data, pd.Series): # Series类型 - 直接转为字典 data = gua_data.to_dict() elif isinstance(gua_data, dict): # 已经是字典类型 data = gua_data else: # 未知类型返回空字典 return {} # 统一处理字典数据 return { 'id': int(data.get('id', 0)), 'trigram': str(data.get('trigram', '')), 'yaoAll': str(data.get('yaoAll', '')), 'yao1': int(data.get('yao1', 0)), 'yao2': int(data.get('yao2', 0)), 'yao3': int(data.get('yao3', 0)), 'yao4': int(data.get('yao4', 0)), 'yao5': int(data.get('yao5', 0)), 'yao6': int(data.get('yao6', 0)), 'value_2binary': int(data.get('value_2binary', 0)), 'change_omit': int(data.get('change_omit', 0)), 'type': str(data.get('Type', 'unknown')) } # 构建完整结果 build_start = time.time() result = { 'date': kDate.strftime('%Y-%m-%d'), 'year_gua': { **format_gua_data(Gua1Year), **get_piwen_info(format_gua_data(Gua1Year).get('trigram', '')) }, 'month_gua': { **format_gua_data(Gua1Month), **get_piwen_info(format_gua_data(Gua1Month).get('trigram', '')) }, 'day_gua': { **format_gua_data(Gua1Day), **get_piwen_info(format_gua_data(Gua1Day).get('trigram', '')) }, 'hour_gua': { **format_gua_data(Gua1Hour), **get_piwen_info(format_gua_data(Gua1Hour).get('trigram', '')) }, 'luck_gua': { **format_gua_data(GuaLuck), **get_piwen_info(format_gua_data(GuaLuck).get('trigram', '')) }, #年天时 'luck_tianshi_years':[ { 'year': row['Year'], 'trigram': row['Trigram'], **get_piwen_info(row['Trigram']) } for _, row in yearGuaMap.iterrows() ], #年人运 'luck_years': [ { 'year': row['Year'], 'trigram': row['trigram'], **get_piwen_info(row['trigram']) } for _, row in LuckYear.iterrows() ] if isinstance(LuckYear, pd.DataFrame) else [], #月天时 'luck_tianshi_month':[ { 'month': row['month'], 'trigram': row['Trigram'], **get_piwen_info(row['Trigram']) } for _, row in yearGuaMap2.iterrows() ], #月人运 'luck_years_month': [ { 'month': row['month'], 'trigram': row['trigram'], **get_piwen_info(row['trigram']) } for _, row in LuckYear_month.iterrows() ] if isinstance(LuckYear, pd.DataFrame) else [], # # 日天时 # 'luck_tianshi_days': [ # { # 'date': row['date'].strftime('%Y-%m-%d'), # 'trigram': row['Trigram'], # **get_piwen_info(row['Trigram']) # } for _, row in yearGuaMap3.iterrows() # ], # # 日人运 # 'luck_days': [ # { # 'date': row['date'].strftime('%Y-%m-%d'), # 'trigram': row['trigram'], # **get_piwen_info(row['trigram']) # } for _, row in LuckYear_day.iterrows() # ] if isinstance(LuckYear_day, pd.DataFrame) else [] } logger.info(f"年卦数据: {format_gua_data(Gua1Year)}") logger.info(f"月卦数据: {format_gua_data(Gua1Month)}") logger.info(f"日卦数据: {format_gua_data(Gua1Day)}") logger.info(f"时卦数据: {format_gua_data(Gua1Hour)}") logger.info(f"吉凶卦数据: {format_gua_data(GuaLuck)}") logger.info(f"结果构建完成，耗时: {time.time() - build_start:.2f}秒") # 在打印 JSON 前检查内容 print("=== 要输出的 JSON 内容 ===", file=sys.stderr) print(result, file=sys.stderr) logger.info("result:",result) # 输出 JSON 格式的结果 # print(json.dumps(result, ensure_ascii=False, indent=4)) # 添加 indent 参数更好查看格式 # 输出结果 output_start = time.time() print(json.dumps(result, ensure_ascii=False, indent=2)) logger.info(f"结果输出完成，耗时: {time.time() - output_start:.2f}秒") total_time = time.time() - total_start logger.info(f"=== API 执行完成，总耗时: {total_time:.2f}秒 ===") except Exception as e: import traceback traceback.print_exc(file=sys.stderr) # 打印错误堆栈，帮助调试 print(f"计算错误: {str(e)}", file=sys.stderr) sys.exit(1)