commit 2bb749800995084c721b371d0bb072da61c1f848 Author: wangzhiming <1115712903@qq.com> Date: Thu May 22 16:59:44 2025 +0800 上传文件至 / diff --git a/API.py b/API.py new file mode 100644 index 0000000..c6e67f0 --- /dev/null +++ b/API.py @@ -0,0 +1,513 @@ + +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) \ No newline at end of file diff --git a/bianyao_huangjijingshi.py b/bianyao_huangjijingshi.py new file mode 100644 index 0000000..e83fc33 --- /dev/null +++ b/bianyao_huangjijingshi.py @@ -0,0 +1,56 @@ +import pandas as pd + +def bianyao_huangjijingshi(Map64Gua: pd.DataFrame, yaoOrig: str) -> pd.DataFrame: + """ + 皇极经世：变爻计算 + 输入原爻（如 '000100'），输出所有变爻之后的卦象合集 + """ + #print("Map64Gua:",Map64Gua['yaoAll'].apply(type).value_counts()) + # 找到原始卦所在的行 + if 'yaoAll' not in Map64Gua.columns: + raise ValueError("'yaoAll' 列在 Map64Gua 中不存在，请检查数据") + + Map64Gua = Map64Gua.copy() + Map64Gua['yaoAll'] = Map64Gua['yaoAll'].astype(str).str.zfill(6).str.strip() + yaoOrig = str(yaoOrig).zfill(6).strip() + + # 找到原始卦所在的行 + matches = Map64Gua[Map64Gua['yaoAll'] == yaoOrig] + if matches.empty: + raise ValueError(f"未找到 yaoAll 为 {yaoOrig} 的卦象") + + k2 = matches.iloc[0] # 取第一行 + + # 计算原卦的二进制值 + a0 = (k2['yao1'] + k2['yao2']*2 + k2['yao3']*4 + + k2['yao4']*8 + k2['yao5']*16 + k2['yao6']*32) + + # 依次翻转每一爻，得到新卦的二进制值 + a1 = (k2['yao1'] + k2['yao2']*2 + k2['yao3']*4 + + k2['yao4']*8 + k2['yao5']*16 + (1 - k2['yao6'])*32) + + a2 = (k2['yao1'] + k2['yao2']*2 + k2['yao3']*4 + + k2['yao4']*8 + (1 - k2['yao5'])*16 + k2['yao6']*32) + + a3 = (k2['yao1'] + k2['yao2']*2 + k2['yao3']*4 + + (1 - k2['yao4'])*8 + k2['yao5']*16 + k2['yao6']*32) + + a4 = (k2['yao1'] + k2['yao2']*2 + (1 - k2['yao3'])*4 + + k2['yao4']*8 + k2['yao5']*16 + k2['yao6']*32) + + a5 = (k2['yao1'] + (1 - k2['yao2'])*2 + k2['yao3']*4 + + k2['yao4']*8 + k2['yao5']*16 + k2['yao6']*32) + + a6 = ((1 - k2['yao1']) + k2['yao2']*2 + k2['yao3']*4 + + k2['yao4']*8 + k2['yao5']*16 + k2['yao6']*32) + + # 结果列表 + result = pd.DataFrame() + + # 查找对应的卦象 + for a in [a0, a1, a2, a3, a4, a5, a6]: + row = Map64Gua[Map64Gua['value_2binary'] == a] + if not row.empty: + result = pd.concat([result, row], ignore_index=True) + + return result diff --git a/guaCalc_huangjijingshi.py b/guaCalc_huangjijingshi.py new file mode 100644 index 0000000..8b4cf49 --- /dev/null +++ b/guaCalc_huangjijingshi.py @@ -0,0 +1,128 @@ +import pandas as pd +import numpy as np +from datetime import datetime, timedelta + +def guaCalc_huangjijingshi(Map64Gua, Map24Jieqi, kDate): + """ + Calculate year, month, day, and hour gua based on Huangji Jingshi method + Args: + Map64Gua (pd.DataFrame): 64 gua mapping table + Map24Jieqi (pd.DataFrame): 24 solar terms data + kDate (datetime): Target date for calculation + Returns: + tuple: (Gua4Hour, Gua1Day, Gua1Month, Gua1Year, GuaLuck) + """ + + # Helper functions that need to be implemented separately + from bianyao_huangjijingshi import bianyao_huangjijingshi + from tuigua_huangjijingshi import tuigua_huangjijingshi + + # Bagua for 360 years + yearRef = {'trigram': '姤', 'value_2binary': '111110', 'start_year': 1744, 'end_year': 2103} + Gua360Year = bianyao_huangjijingshi(Map64Gua, yearRef['value_2binary']) + Gua360Year['yearStart'] = [yearRef['start_year'] + i * 60 for i in range(7)] + Gua360Year['yearStart'] = pd.concat([pd.Series([yearRef['start_year']]), Gua360Year['yearStart']]).reset_index(drop=True) + Gua360Year['yearEnd'] = [start + 59 for start in Gua360Year['yearStart']] + Gua360Year.loc[0, 'yearEnd'] = yearRef['end_year'] + + # Bagua for 60 years + kYear = kDate.year + kYearOrig = kYear + + solar_terms = Map24Jieqi + solar_terms1 = solar_terms[solar_terms['Solar_Terms'] == '冬至'] + + # Find the last winter solstice before our date + k1 = np.where((kDate - solar_terms1['As_Of_Date']) > pd.Timedelta(0))[0][-1] + kYear = (solar_terms1['As_Of_Date'].iloc[k1] + pd.Timedelta(days=90)).year + + # Find which 60-year period we're in + mask = ((Gua360Year['yearStart'] <= kYear) & (Gua360Year['yearEnd'] >= kYear)) + k1 = np.where(mask)[0] + if len(k1) > 1: + k1 = k1[1:] + k1 = k1[0] + + a1 = Gua360Year['value_2binary'].iloc[k1] + a2 = kYear - Gua360Year['yearStart'].iloc[k1] + Gua1Year = tuigua_huangjijingshi(Map64Gua, a1, a2 + 1) + Gua1Year['Type'] = 'year' + + # Get the yaoAll value correctly + if isinstance(Gua1Year, pd.DataFrame): + trigram_value = Gua1Year['trigram'].iloc[0] + yaoAll_value = Gua1Year['yaoAll'].iloc[0] + else: + trigram_value = Gua1Year['trigram'] + yaoAll_value = Gua1Year['yaoAll'] + + # Bagua for 360 days (months) + dayRef = { + 'trigram': trigram_value, + 'yaoAll': yaoAll_value, + 'start_day': 1, + 'end_day': 360 + } + + Gua365days = bianyao_huangjijingshi(Map64Gua, str(yaoAll_value)) + + # Find important solar terms + solar_termsSimple = Map24Jieqi[Map24Jieqi['isImportant'] == 1] + + # Find the last important solar term before our date + kk = np.where((kDate - solar_termsSimple['As_Of_Date']) > pd.Timedelta(0))[0][-1] + + # Find winter solstice within these terms + kk2 = np.where(solar_termsSimple['Solar_Terms'].iloc[:kk+1] == '冬至')[0][-1] + + a1 = Gua365days['value_2binary'].iloc[kk - kk2 + 1] + + Gua1Month = Gua365days.iloc[[kk - kk2 + 1]].copy() + Gua1Month['Type'] = 'month' + + # Calculate day gua + kDateMonth = solar_termsSimple['As_Of_Date'].iloc[kk] + kGap = (kDate.date() - kDateMonth.date()).days + a2 = round(kGap) + Gua1Day = tuigua_huangjijingshi(Map64Gua, a1, a2) + Gua1Day['Type'] = 'day' + + # Calculate hour gua (4-hour blocks) + # Get yaoAll value from Gua1Day - handle both DataFrame and scalar cases + if isinstance(Gua1Day, pd.DataFrame): + day_yaoAll = Gua1Day['yaoAll'].iloc[0] + else: + day_yaoAll = Gua1Day['yaoAll'] + + Gua24Hours = bianyao_huangjijingshi(Map64Gua, str(day_yaoAll)) + k1 = min(int(np.floor(kDate.hour / 4))+1, len(Gua24Hours) - 1) #ceil修改为了floor + Gua4Hour = Gua24Hours.iloc[[k1]].copy() + Gua4Hour['Type'] = 'hour' + def to_single_row_df(x): + if isinstance(x, pd.Series): + return pd.DataFrame([x])[['yao1', 'yao2', 'yao3', 'yao4', 'yao5', 'yao6']] + elif isinstance(x, pd.DataFrame): + return x[['yao1', 'yao2', 'yao3', 'yao4', 'yao5', 'yao6']].reset_index(drop=True) + else: + raise TypeError("Unsupported type: must be Series or DataFrame") + # Calculate luck gua + # abc = pd.concat([ + # Gua1Year[['yao1', 'yao2', 'yao3', 'yao4', 'yao5', 'yao6']], + # Gua1Month[['yao1', 'yao2', 'yao3', 'yao4', 'yao5', 'yao6']], + # Gua1Day[['yao1', 'yao2', 'yao3', 'yao4', 'yao5', 'yao6']], + # Gua4Hour[['yao1', 'yao2', 'yao3', 'yao4', 'yao5', 'yao6']] + # ]) + abc = pd.concat([ + to_single_row_df(Gua1Year), + to_single_row_df(Gua1Month), + to_single_row_df(Gua1Day), + to_single_row_df(Gua4Hour) + ], ignore_index=True) + abc2 = np.mod(np.nansum(abc, axis=0), 2)[:6] + abc3 = np.sum(abc2 * [1, 2, 4, 8, 16, 32]) + k1 = np.where(Map64Gua['value_2binary'] == abc3)[0][0] + GuaLuck = Map64Gua.iloc[[k1]].copy() + GuaLuck['Type'] = 'luck' + # print(abc.dtypes) + # print(abc) + return Gua4Hour, Gua1Day, Gua1Month, Gua1Year, GuaLuck \ No newline at end of file diff --git a/huangjijingshi20250217.py b/huangjijingshi20250217.py new file mode 100644 index 0000000..c22a2aa --- /dev/null +++ b/huangjijingshi20250217.py @@ -0,0 +1,88 @@ +import pandas as pd +import numpy as np +from datetime import datetime +import pymysql +from sqlalchemy import create_engine + +# 导入自定义卦象计算模块 +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 +engine = create_engine(f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}') + +# ========== 文件路径配置 ========== +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' + df = pd.read_sql(sqlquery, engine) + df['As_Of_Date'] = pd.to_datetime(df['As_Of_Date']) + df = df.sort_values('As_Of_Date') + return df + + +# 加载数据并标记重要节气 +solar_terms = load_solar_terms(engine) +important_terms = ['冬至', '雨水', '谷雨', '夏至', '处暑', '霜降'] +solar_terms['isImportant'] = solar_terms['Solar_Terms'].isin(important_terms).astype(int) +Map24Jieqi = solar_terms.copy() + +# ========== 示例：计算某个日期的卦象 ========== +kDate = datetime(1973, 7, 17, 17, 20, 0) +print("计算日期:", kDate) + +# 调用卦象计算函数 +Gua1Hour, Gua1Day, Gua1Month, Gua1Year, GuaLuck = guaCalc_huangjijingshi( + Map64Gua, Map24Jieqi, kDate +) + +print("年卦:", Gua1Year) +print("月卦:", Gua1Month) +print("日卦:", Gua1Day) +print("时卦:", Gua1Hour) +print("吉凶卦:", GuaLuck) + +# ========== 计算 1990-2030 每年的年卦 ========== +# 生成日期范围 +start_date = datetime(1990, 1, 1) +end_date = datetime(2030, 1, 1) +date_range = pd.date_range(start=start_date, end=end_date, freq='YS') # 每年第一天 + +year_gua_list = [] +for date in date_range: + _, _, _, Gua1Year, _ = guaCalc_huangjijingshi(Map64Gua, Map24Jieqi, date) + year_gua_list.append({ + 'Year': date.year, + 'Trigram': Gua1Year.trigram if hasattr(Gua1Year, 'trigram') else None, + }) + +yearGuaMap = pd.DataFrame(year_gua_list) + +#========== 计算年份吉凶 ========== +LuckYear = luckCalc_huangjijingshi(Map64Gua, yearGuaMap, GuaLuck) +print("\n年份吉凶计算结果:") +print(LuckYear) \ No newline at end of file diff --git a/luckCalc_huangjijingshi.py b/luckCalc_huangjijingshi.py new file mode 100644 index 0000000..b4625b5 --- /dev/null +++ b/luckCalc_huangjijingshi.py @@ -0,0 +1,51 @@ +import pandas as pd +import numpy as np + +def luckCalc_huangjijingshi(Map64Gua, yearGuaMap, GuaLuck): + """ + 计算年份吉凶卦象 + Args: + Map64Gua (pd.DataFrame): 64卦映射表，包含卦名、爻信息和二进制值 + yearGuaMap (list or pd.DataFrame): 年份卦象映射表，格式为 [年份, 卦名] + GuaLuck (pd.DataFrame): 吉凶卦象，包含6爻信息 + Returns: + pd.DataFrame: 包含年份、年卦和吉凶卦的结果表 + """ + LuckYear = [] + + # 确保 yearGuaMap 是 DataFrame（如果是列表，先转换） + if isinstance(yearGuaMap, list): + yearGuaMap = pd.DataFrame(yearGuaMap, columns=['Year', 'Trigram']) + + for i in range(len(yearGuaMap)): + kYear = yearGuaMap.iloc[i] # 当前年份和卦名 + trigram = kYear['Trigram'] + + # 找到当前卦名对应的卦象（爻信息） + k1 = Map64Gua[Map64Gua['trigram'] == trigram].index + if len(k1) == 0: + raise ValueError(f"未找到卦名 {trigram} 对应的卦象") + + # 提取当前卦的6爻和吉凶卦的6爻，合并计算 + abc = Map64Gua.loc[k1, ['yao1', 'yao2', 'yao3', 'yao4', 'yao5', 'yao6']].values + abc_luck = GuaLuck[['yao1', 'yao2', 'yao3', 'yao4', 'yao5', 'yao6']].values + abc_combined = np.vstack([abc, abc_luck]) + + # 计算新卦的二进制值（模2求和后加权） + abc2 = np.mod(np.nansum(abc_combined, axis=0), 2) + abc3 = np.sum(abc2 * [1, 2, 4, 8, 16, 32]) + + # 查找对应的新卦 + new_gua = Map64Gua[Map64Gua['value_2binary'] == abc3].iloc[0] + LuckYear.append(new_gua) + + # 合并结果 + LuckYear = pd.DataFrame(LuckYear) + LuckYear['Year'] = yearGuaMap['Year'].values + LuckYear['YearGua'] = yearGuaMap['Trigram'].values + + # 调整列顺序（年份和年卦在前） + cols = ['Year', 'YearGua'] + [c for c in LuckYear.columns if c not in ['Year', 'YearGua']] + LuckYear = LuckYear[cols] + + return LuckYear \ No newline at end of file