update

utils/__init__.py

@@ -1,9 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-工具模块 - 包含各种工具函数
-"""
-
-from .file_utils import load_alpha_list, save_results_to_file
-from .time_utils import format_time
-
-__all__ = ['load_alpha_list', 'save_results_to_file', 'format_time']

utils/file_utils.py

@@ -1,84 +0,0 @@
-# -*- coding: utf-8 -*-
-import os
-import json
-import time
-from typing import List, Any
-
-
-def load_alpha_list(file_path: str) -> List[str]:
-    """从文件加载Alpha因子列表"""
-    if not os.path.exists(file_path):
-        print(f"{file_path} 文件不存在")
-        with open(file_path, 'w', encoding='utf-8') as file:
-            file.write("")
-        print(f"已创建 {file_path} 文件, 请添加因子后重新运行, 一行一个因子")
-        return []
-
-    with open(file_path, 'r', encoding='utf-8') as file:
-        alpha_list = [line.strip() for line in file if line.strip()]
-
-    return alpha_list
-
-
-def save_results_to_file(results: List[Any], result_dir: str = 'result') -> str:
-    """保存结果到文件"""
-    # 转换为可序列化的格式
-    serializable_results = []
-    for result in results:
-        if hasattr(result, '__dict__'):
-            # 如果是dataclass对象，转换为字典
-            result_dict = result.__dict__.copy()
-        else:
-            # 如果是字典
-            result_dict = result.copy()
-
-        # 处理时间消耗
-        if 'time_consuming' in result_dict:
-            result_dict['time_consuming'] = round(result_dict['time_consuming'], 2)
-
-        # 只需要确保所有浮点数精度一致
-        def round_floats(obj):
-            if isinstance(obj, float):
-                return round(obj, 6)
-            elif isinstance(obj, dict):
-                return {k: round_floats(v) for k, v in obj.items()}
-            elif isinstance(obj, list):
-                return [round_floats(item) for item in obj]
-            else:
-                return obj
-
-        # 对结果中的所有浮点数进行精度处理
-        result_dict = round_floats(result_dict)
-
-        serializable_results.append(result_dict)
-
-    # 确保结果目录存在
-    if not os.path.exists(result_dir):
-        os.makedirs(result_dir)
-
-    result_name = f"{result_dir}/simulation_results-{str(int(time.time()))}.json"
-    with open(result_name, 'w', encoding='utf-8') as f:
-        json.dump(serializable_results, f, ensure_ascii=False, indent=2)
-
-    print(f"结果已保存到 {result_name}")
-    return result_name
-
-
-def save_success_alpha(success_expression_list):
-    """
-    将成功的表达式列表保存到 success.txt 文件中
-    如果文件不存在则创建，如果存在则追加写入
-    """
-    if not success_expression_list:
-        print("没有成功的表达式需要保存")
-        return
-
-    try:
-        with open('success.txt', 'a', encoding='utf-8') as f:
-            for expression in success_expression_list:
-                f.write(expression + '\n')
-
-        print(f"成功保存 {len(success_expression_list)} 个表达式到 success.txt")
-
-    except Exception as e:
-        print(f"保存文件时出错: {e}")

utils/helpers.py

@@ -0,0 +1,23 @@
+# -*- coding: utf-8 -*-
+"""工具函数"""
+
+import time
+from typing import Callable, Any
+
+
+def retry_on_exception(max_retries: int = 3, delay: float = 5.0) -> Callable:
+    """异常重试装饰器"""
+    def decorator(func: Callable) -> Callable:
+        def wrapper(*args, **kwargs) -> Any:
+            for attempt in range(max_retries):
+                try:
+                    return func(*args, **kwargs)
+                except Exception as e:
+                    if attempt == max_retries - 1:
+                        raise e
+                    print(f"尝试 {attempt + 1}/{max_retries} 失败: {e}")
+                    print(f"{delay}秒后重试...")
+                    time.sleep(delay)
+            return None
+        return wrapper
+    return decorator

utils/time_utils.py

@@ -1,9 +0,0 @@
-# -*- coding: utf-8 -*-
-def format_time(seconds: float) -> str:
-    """将秒数格式化为 xx分xx秒 格式"""
-    if seconds < 60:
-        return f"{seconds:.2f}秒"
-    else:
-        minutes = int(seconds // 60)
-        remaining_seconds = seconds % 60
-        return f"{minutes}分{remaining_seconds:.2f}秒"