上传文件至 /
This commit is contained in:
commit
2bb7498009
513
API.py
Normal file
513
API.py
Normal file
@ -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)
|
||||
56
bianyao_huangjijingshi.py
Normal file
56
bianyao_huangjijingshi.py
Normal file
@ -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
|
||||
128
guaCalc_huangjijingshi.py
Normal file
128
guaCalc_huangjijingshi.py
Normal file
@ -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
|
||||
88
huangjijingshi20250217.py
Normal file
88
huangjijingshi20250217.py
Normal file
@ -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)
|
||||
51
luckCalc_huangjijingshi.py
Normal file
51
luckCalc_huangjijingshi.py
Normal file
@ -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
|
||||
Loading…
x
Reference in New Issue
Block a user