diff --git a/alpha-forge/simple_alpha_fetcher.py b/alpha-forge/scripts/01_fetch_alphas.py similarity index 100% rename from alpha-forge/simple_alpha_fetcher.py rename to alpha-forge/scripts/01_fetch_alphas.py diff --git a/alpha-forge/feature_analysis_database.py b/alpha-forge/scripts/02_init_db.py similarity index 100% rename from alpha-forge/feature_analysis_database.py rename to alpha-forge/scripts/02_init_db.py diff --git a/alpha-forge/simple_frature_analysis.py b/alpha-forge/scripts/03_import_and_extract.py similarity index 100% rename from alpha-forge/simple_frature_analysis.py rename to alpha-forge/scripts/03_import_and_extract.py diff --git a/alpha-forge/feature_attribution.py b/alpha-forge/scripts/04_calc_gain.py similarity index 100% rename from alpha-forge/feature_attribution.py rename to alpha-forge/scripts/04_calc_gain.py diff --git a/alpha-forge/scripts/05_test_extractor.py b/alpha-forge/scripts/05_test_extractor.py new file mode 100644 index 0000000..612b630 --- /dev/null +++ b/alpha-forge/scripts/05_test_extractor.py @@ -0,0 +1,62 @@ +# -*- coding: utf-8 -*- +import re +import pandas as pd +import json +import ast # Python 的 ast.literal_eval 可以安全解析 Python 字典字符串 + +def test_code_001(): + df = pd.read_csv("alpha_list.csv", nrows=5) + + # 方法1：用 ast.literal_eval（推荐，能解析 Python 字面量） + for i, row in df.iterrows(): + regular_str = row['regular'] + if pd.notna(regular_str): + try: + regular_dict = ast.literal_eval(regular_str) + code = regular_dict.get('code', '') + print(f"Row {i}: {code[:80]}...") + except Exception as e: + print(f"Row {i}: 解析失败 - {e}") + + # 方法2：看看原始字符串前200个字符 + print("\n原始字符串示例:") + print(df.iloc[0]['regular'][:200]) + +def test_code_002(): + expression = "last_diff_value(ts_sum(subtract(implied_volatility_call_120, implied_volatility_put_90), 20), 5)" + + # 提取算子 + operators = set() + for match in re.finditer(r'\b([a-z_][a-z0-9_]*)\s*\(', expression): + op = match.group(1) + operators.add(op) + print(f"匹配到算子: {op}") + + print(f"\n所有算子: {operators}") + + # 提取字段 + candidates = set(re.findall(r'\b([a-z][a-z0-9_]*)\b', expression)) + print(f"\n所有候选词: {candidates}") + + # 过滤掉算子 + fields = [c for c in candidates if c not in operators and not c.isdigit() and len(c) > 2] + print(f"\n字段: {fields}") + +def test_code_003(): + df = pd.read_csv("alpha_list.csv") + + # 检查 id 列的类型 + print(f"id 列类型: {df['id'].dtype}") + + # 检查是否有重复 + duplicates = df[df.duplicated(subset=['id'], keep=False)] + print(f"\n重复的 id 数量: {len(duplicates)}") + + if len(duplicates) > 0: + print("\n重复的 id 示例:") + for aid in duplicates['id'].unique()[:5]: + rows = df[df['id'] == aid] + print(f" {aid}: 出现 {len(rows)} 次") + # 打印这些行的 fitness 看是否相同 + for _, row in rows.iterrows(): + print(f" fitness: {row['is'][:100] if isinstance(row['is'], str) else row['is']}...") \ No newline at end of file diff --git a/alpha-forge/test_code_001.py b/alpha-forge/test_code_001.py deleted file mode 100644 index bd4a810..0000000 --- a/alpha-forge/test_code_001.py +++ /dev/null @@ -1,20 +0,0 @@ -import pandas as pd -import json -import ast # Python 的 ast.literal_eval 可以安全解析 Python 字典字符串 - -df = pd.read_csv("alpha_list.csv", nrows=5) - -# 方法1：用 ast.literal_eval（推荐，能解析 Python 字面量） -for i, row in df.iterrows(): - regular_str = row['regular'] - if pd.notna(regular_str): - try: - regular_dict = ast.literal_eval(regular_str) - code = regular_dict.get('code', '') - print(f"Row {i}: {code[:80]}...") - except Exception as e: - print(f"Row {i}: 解析失败 - {e}") - -# 方法2：看看原始字符串前200个字符 -print("\n原始字符串示例:") -print(df.iloc[0]['regular'][:200]) \ No newline at end of file diff --git a/alpha-forge/test_code_002.py b/alpha-forge/test_code_002.py deleted file mode 100644 index fb80784..0000000 --- a/alpha-forge/test_code_002.py +++ /dev/null @@ -1,20 +0,0 @@ -import re - -expression = "last_diff_value(ts_sum(subtract(implied_volatility_call_120, implied_volatility_put_90), 20), 5)" - -# 提取算子 -operators = set() -for match in re.finditer(r'\b([a-z_][a-z0-9_]*)\s*\(', expression): - op = match.group(1) - operators.add(op) - print(f"匹配到算子: {op}") - -print(f"\n所有算子: {operators}") - -# 提取字段 -candidates = set(re.findall(r'\b([a-z][a-z0-9_]*)\b', expression)) -print(f"\n所有候选词: {candidates}") - -# 过滤掉算子 -fields = [c for c in candidates if c not in operators and not c.isdigit() and len(c) > 2] -print(f"\n字段: {fields}") \ No newline at end of file diff --git a/alpha-forge/test_code_003.py b/alpha-forge/test_code_003.py deleted file mode 100644 index a990457..0000000 --- a/alpha-forge/test_code_003.py +++ /dev/null @@ -1,19 +0,0 @@ -import pandas as pd - -df = pd.read_csv("alpha_list.csv") - -# 检查 id 列的类型 -print(f"id 列类型: {df['id'].dtype}") - -# 检查是否有重复 -duplicates = df[df.duplicated(subset=['id'], keep=False)] -print(f"\n重复的 id 数量: {len(duplicates)}") - -if len(duplicates) > 0: - print("\n重复的 id 示例:") - for aid in duplicates['id'].unique()[:5]: - rows = df[df['id'] == aid] - print(f" {aid}: 出现 {len(rows)} 次") - # 打印这些行的 fitness 看是否相同 - for _, row in rows.iterrows(): - print(f" fitness: {row['is'][:100] if isinstance(row['is'], str) else row['is']}...") \ No newline at end of file diff --git a/alpha-forge/wqb-get-alphas.py b/alpha-forge/wqb-get-alphas.py deleted file mode 100644 index 4455605..0000000 --- a/alpha-forge/wqb-get-alphas.py +++ /dev/null @@ -1,381 +0,0 @@ -import httpx -from httpx import BasicAuth, Timeout -import pandas as pd -from tqdm import tqdm -from pathlib import Path -import logging -from typing import List, Optional -import time - -class AlphaManager: - def __init__(self, credentials_file='account.txt', base_path=None): - """ - 初始化Alpha管理器 - - Args: - credentials_file: 账号文件路径（备用） - base_path: 输出文件保存路径 - """ - self.client = None - self.base_path = Path(base_path) if base_path else Path.cwd() - self.logger = self._setup_logger() - - # 登录 - self.login(credentials_file) - - def _setup_logger(self): - """设置日志记录器""" - logger = logging.getLogger(__name__) - logger.setLevel(logging.INFO) - - if not logger.handlers: - handler = logging.StreamHandler() - formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') - handler.setFormatter(formatter) - logger.addHandler(handler) - - return logger - - def login(self, credentials_file='account.txt'): - """登录WorldQuant Brain API""" - try: - # 从nacos获取账号密码 - with httpx.Client(timeout=10.0) as temp_client: - nacos_resp = temp_client.get( - 'http://192.168.31.41:30848/nacos/v1/cs/configs?dataId=wq_account&group=quantify' - ) - - if nacos_resp.status_code != 200: - self.logger.error('获取账号密码失败') - return False - - config = nacos_resp.json() - username = config.get('user_name') - password = config.get('password') - - if not username or not password: - self.logger.error('账号密码不完整') - return False - - self.logger.info(f"正在登录账户: {username}") - - # 创建客户端并设置超时（关键修复） - # 设置更长的超时时间：连接30秒，读取60秒 - timeout = Timeout(connect=30.0, read=60.0, write=30.0, pool=30.0) - self.client = httpx.Client( - auth=BasicAuth(username, password), - timeout=timeout - ) - - # 发送登录请求 - response = self.client.post('https://api.worldquantbrain.com/authentication') - - if response.status_code == 201: - self.logger.info("登录成功!") - return True - else: - self.logger.error(f"登录失败: {response.status_code} - {response.text}") - self.client.close() - self.client = None - return False - - except Exception as e: - self.logger.error(f"登录异常: {e}") - return False - - def update_alpha_color(self, alpha_id: str, color: str) -> bool: - """标记Alpha颜色""" - if not self.client: - self.logger.error("客户端未登录") - return False - - try: - update_data = {"color": color} - response = self.client.patch( - f"https://api.worldquantbrain.com/alphas/{alpha_id}", - json=update_data - ) - return response.status_code == 200 - except Exception as e: - self.logger.error(f"标记颜色失败: {e}") - return False - - def wechat_check_corr_message(self, message: str): - """微信通知（可选功能）""" - self.logger.info(f"通知消息: {message}") - - def _make_request_with_retry(self, url: str, max_retries: int = 3, retry_delay: float = 2.0): - """ - 带重试机制的请求 - - Args: - url: 请求URL - max_retries: 最大重试次数 - retry_delay: 重试延迟（秒） - - Returns: - httpx.Response 或 None - """ - for attempt in range(max_retries): - try: - response = self.client.get(url) - return response - except Exception as e: - self.logger.warning(f"请求失败 (尝试 {attempt+1}/{max_retries}): {e}") - if attempt < max_retries - 1: - time.sleep(retry_delay) - else: - self.logger.error(f"请求最终失败: {url}") - raise - return None - - def get_alphas( - self, - total_alphas: int = 5, - limit: int = 100, - delay: int = 1, - region: str = "USA", - universe: str = "TOP3000", - required_fields: Optional[List[str]] = None, - match_mode: str = "all", - min_sharpe: Optional[float] = None, - min_fitness: Optional[float] = None, - hidden: str = "false", - submittable: bool = False, - auto_color: bool = False, - color: str = "GREEN", - output_file_name: str = "alpha_search_list.csv", - mode: str = "w", - max_retries: int = 3, - ): - """ - 搜索Alpha并筛选 - - Args: - total_alphas: 最多获取多少个Alpha - limit: 每次API请求获取的数量 - delay: API请求的延迟设置 - region: 市场区域，如 "USA", "CHINA" - universe: 股票池，如 "TOP3000" - required_fields: 关键词列表，如 ['put', 'call'] - match_mode: 关键词匹配模式："all"(全匹配) 或 "any"(任一匹配) - min_sharpe: 最小夏普比率阈值 - min_fitness: 最小适应度阈值 - hidden: 是否搜索已隐藏的Alpha ("true"/"false") - submittable: 是否只筛选可提交（无FAIL检查项）的Alpha - auto_color: 是否自动给符合条件的Alpha标记颜色 - color: 标记的颜色 - output_file_name: 输出的CSV文件名 - mode: 写入模式："w"(覆盖写入) 或 "a"(追加写入) - max_retries: 请求失败时的最大重试次数 - - Returns: - list: 符合条件的Alpha列表 - """ - if not self.client: - self.logger.error("客户端未登录，无法执行搜索") - return [] - - # 验证颜色参数 - valid_colors = [None, "GREEN", "YELLOW", "RED", "BLUE", "PURPLE", "ORANGE"] - if color not in valid_colors: - raise ValueError(f"颜色必须是以下之一: {valid_colors}") - - fetched_alphas = [] - offset = 0 - total_accessed = 0 - colored_count = 0 - - # 先获取总数（带重试） - count_url = ( - f"https://api.worldquantbrain.com/users/self/alphas?stage=IS&hidden={hidden}" - f"&limit=1&settings.delay={delay}&settings.region={region}&status=UNSUBMITTED%1FIS_FAIL&settings.universe={universe}" - ) - - total_available = 0 - for attempt in range(max_retries): - try: - count_response = self.client.get(count_url) - total_available = count_response.json()["count"] - break - except Exception as e: - self.logger.warning(f"获取Alpha总数失败 (尝试 {attempt+1}/{max_retries}): {e}") - if attempt < max_retries - 1: - time.sleep(2.0) - else: - self.logger.error("获取Alpha总数最终失败") - return [] - - if total_available == 0: - self.logger.warning("未找到任何Alpha") - return [] - - self.logger.info(f"共找到 {total_available} 个Alpha，开始筛选...") - - pbar = tqdm(total=min(total_available, 10000), desc="扫描Alpha", unit="条") - - while len(fetched_alphas) < total_alphas and offset < total_available: - # 构建请求URL - url = ( - f"https://api.worldquantbrain.com/users/self/alphas?stage=IS&limit={limit}" - f"&offset={offset}&settings.delay={delay}&settings.region={region}&hidden={hidden}&status=UNSUBMITTED%1FIS_FAIL&settings.universe={universe}" - ) - - try: - # 使用重试机制 - response = None - for attempt in range(max_retries): - try: - response = self.client.get(url) - break - except Exception as e: - self.logger.warning(f"请求失败 (尝试 {attempt+1}/{max_retries}): {e}") - if attempt < max_retries - 1: - time.sleep(2.0) - else: - raise - - if response.status_code == 400: - self.logger.warning(f"遇到API限制 (offset={offset})，停止获取更多数据") - break - - response_data = response.json() - - if not isinstance(response_data, dict) or "results" not in response_data: - self.logger.error(f"API返回了意外的数据: {response_data}") - break - - alphas = response_data["results"] - - if not alphas: - break - - total_accessed += len(alphas) - - # 关键词筛选 - if required_fields: - if match_mode == "all": - filtered_alphas = [ - alpha for alpha in alphas - if all(field in alpha["regular"]["code"] for field in required_fields) - ] - elif match_mode == "any": - filtered_alphas = [ - alpha for alpha in alphas - if any(field in alpha["regular"]["code"] for field in required_fields) - ] - else: - raise ValueError("match_mode 必须是 'all' 或 'any'") - else: - filtered_alphas = alphas - - # 进一步筛选夏普比率和fitness - final_filtered = [] - for alpha in filtered_alphas: - sharpe = alpha.get("is", {}).get("sharpe", 0) - fitness = alpha.get("is", {}).get("fitness", 0) - - sharpe_ok = (min_sharpe is None) or (sharpe is not None and abs(sharpe) >= min_sharpe) - fitness_ok = (min_fitness is None) or (fitness is not None and abs(fitness) >= min_fitness) - - if sharpe_ok and fitness_ok: - if submittable: - checks = alpha.get("is", {}).get("checks", []) - fail_count = sum(1 for check in checks if check.get("result") == "FAIL") - - if fail_count == 0 and auto_color: - alpha_id = alpha.get("id") - if alpha_id: - success = self.update_alpha_color(alpha_id, color) - if success: - colored_count += 1 - - if fail_count == 0: - final_filtered.append(alpha) - else: - final_filtered.append(alpha) - - fetched_alphas.extend(final_filtered) - - # 更新进度条 - pbar.update(len(alphas)) - pbar.set_postfix({ - "Region": region, - "已扫描": total_accessed, - "找到": len(fetched_alphas), - "本次": len(final_filtered), - "标记": colored_count, - }) - - if len(alphas) < limit: - break - - offset += limit - - # 添加延迟避免请求过快 - time.sleep(0.5) - - except Exception as e: - self.logger.error(f"请求失败: {e}") - break - - pbar.close() - - # 输出标记统计 - if auto_color and colored_count > 0: - self.logger.info(f"共标记了 {colored_count} 个Alpha颜色为{color}") - self.wechat_check_corr_message(f"共标记了 {colored_count} 个Alpha颜色为{color}") - - alpha_list = fetched_alphas[:total_alphas] - - if not alpha_list: - self.logger.warning("未找到任何符合条件的Alpha！请检查筛选条件是否过于严格。") - else: - df = pd.DataFrame(alpha_list) - df["temp_fitness"] = df.apply( - lambda row: row["is"].get("fitness", 0) if isinstance(row.get("is"), dict) else 0, - axis=1 - ) - df_sorted = df.sort_values(by="temp_fitness", ascending=False) - df_sorted = df_sorted.drop("temp_fitness", axis=1) - - output_path = self.base_path / output_file_name - - if mode == "w": - df_sorted.to_csv(output_path, index=False) - elif mode == "a": - df_sorted.to_csv(output_path, mode="a", index=False, header=False) - - self.logger.info(f"批量回测初步检测结果已经下载！共{len(alpha_list)}条记录！\n{output_path}文件名保存！") - - return alpha_list - - def close(self): - """关闭客户端""" - if self.client: - self.client.close() - - -# 使用示例 -if __name__ == "__main__": - # 创建管理器实例（会自动登录） - manager = AlphaManager() - - if manager.client: - # 搜索包含"put"关键词的可提交Alpha - results = manager.get_alphas( - total_alphas=10, - region="USA", - universe="TOP3000", - required_fields=['put'], - match_mode="all", - submittable=True, - auto_color=True, - color="GREEN", - output_file_name="my_alpha_list.csv" - ) - - print(f"找到 {len(results) if results else 0} 个符合条件的Alpha") - - # 关闭客户端 - manager.close() \ No newline at end of file diff --git a/rpc_alpha_workflow/01_generate_direction.py b/rpc_alpha_workflow/01_generate_direction.py new file mode 100644 index 0000000..b37af0c --- /dev/null +++ b/rpc_alpha_workflow/01_generate_direction.py @@ -0,0 +1,191 @@ +# -*- coding: utf-8 -*- +import random +import time +from xmlrpc.client import ServerProxy, Fault + +is_local = 0 +if is_local: + url = 'http://127.0.0.1:28888' +else: + url = 'http://192.168.31.41:32000' + +username = 'rpc' +password = 'aaaAAA111' +db = "quantify" + +common = ServerProxy(f"{url}/xmlrpc/2/common") +uid = common.authenticate(db, username, password, {}) + +models = ServerProxy(f"{url}/xmlrpc/2/object") + +# 1. 在 alpha.generate.research.direction 表中创建一条数据 +# 需要准备: region_id 和 universe_id + +region = 'USA' +universe = 'TOP3000' + +# 根据名称搜索 region 和 universe 的ID +region_ids = models.execute_kw(db, uid, password, 'alpha.region.settings', 'search', [[['name', '=', region]]], {'limit': 1}) +universe_ids = models.execute_kw(db, uid, password, 'alpha.universe.settings', 'search', [[['name', '=', universe]]], {'limit': 1}) + +if not region_ids: + print(f"没有找到 region: '{region}'，请检查名称是否正确") + exit(1) +if not universe_ids: + print(f"没有找到 universe: '{universe}'，请检查名称是否正确") + exit(1) + +region_id = region_ids[0] +universe_id = universe_ids[0] + +# 获取 region 和 universe 的 name +region_data = models.execute_kw(db, uid, password, 'alpha.region.settings', 'read', [[region_id], ['name']]) +universe_data = models.execute_kw(db, uid, password, 'alpha.universe.settings', 'read', [[universe_id], ['name']]) +region_name = region_data[0]['name'] if region_data else 'Unknown' +universe_name = universe_data[0]['name'] if universe_data else 'Unknown' + +print(f"使用 region_id: {region_id} (name: {region_name}), universe_id: {universe_id} (name: {universe_name})") + +# 随机选择一个 llm_settings_line_id +llm_line_ids = models.execute_kw(db, uid, password, 'llm.settings.line', 'search', [[['id', '!=', 0]]]) +if not llm_line_ids: + print("没有找到 llm.settings.line 记录") + exit(1) + +llm_settings_line_id = random.choice(llm_line_ids) +llm_line_data = models.execute_kw(db, uid, password, 'llm.settings.line', 'read', [[llm_settings_line_id], ['display_name', 'model_name']]) +llm_display_name = llm_line_data[0].get('display_name', 'Unknown') if llm_line_data else 'Unknown' +print(f"随机选择 llm_settings_line_id: {llm_settings_line_id} (name: {llm_display_name})") + +# 搜索 Research Direction Prompt +rd_prompt_ids = models.execute_kw(db, uid, password, 'alpha.prompt.settings', 'search', [[['prompt_type', '=', 'Research Direction']]], {'limit': 1}) +if not rd_prompt_ids: + print("没有找到 prompt_type='Research Direction' 的 alpha.prompt.settings 记录") + exit(1) +research_direction_prompt_id = rd_prompt_ids[0] +print(f"使用 Research Direction Prompt ID: {research_direction_prompt_id}") + +# 搜索 Feature Engineering Prompt +fe_prompt_ids = models.execute_kw(db, uid, password, 'alpha.prompt.settings', 'search', [[['prompt_type', '=', 'Feature Engineering']]], {'limit': 1}) +if not fe_prompt_ids: + print("没有找到 prompt_type='Feature Engineering' 的 alpha.prompt.settings 记录") + exit(1) +feature_engineering_prompt_id = fe_prompt_ids[0] +print(f"使用 Feature Engineering Prompt ID: {feature_engineering_prompt_id}") + +# 创建研究方向记录 +record_vals = { + 'region': region_id, + 'universe': universe_id, + 'llm_settings_line_id': llm_settings_line_id, + 'research_direction_prompt': research_direction_prompt_id, + 'feature_engineering_prompt': feature_engineering_prompt_id, +} + +record_id = models.execute_kw(db, uid, password, 'alpha.generate.research.direction', 'create', [record_vals]) +print(f"成功创建研究方向记录，ID: {record_id}") + +# 2. 调用 btn_quick_selection_datasets 按钮方法 +result = models.execute_kw(db, uid, password, 'alpha.generate.research.direction', 'btn_quick_selection_datasets', [[record_id]]) +print(f"btn_quick_selection_datasets 执行结果: {result}") + +# 验证数据集行是否已创建 +datasets_line_ids = models.execute_kw(db, uid, password, 'alpha.research.datasets.line', 'search', [[['research_direction_id', '=', record_id]]]) +print(f"已创建的数据集行数量: {len(datasets_line_ids)}") + +# 获取数据集行的 dataset_code（数据集名字） +if datasets_line_ids: + datasets_lines = models.execute_kw(db, uid, password, 'alpha.research.datasets.line', 'read', [datasets_line_ids, ['datasets_code']]) + dataset_codes = [line.get('datasets_code') for line in datasets_lines if line.get('datasets_code')] + print(f"关联的数据集名字: {dataset_codes}") + print(f"数据集总数: {len(dataset_codes)}") + + +# 3. 点击 btn_generate_research 按钮 +print("\n开始生成研究方向...") +try: + result = models.execute_kw(db, uid, password, 'alpha.generate.research.direction', 'btn_generate_research', [[record_id]]) + print(f"btn_generate_research 执行结果: {result}") +except Fault as e: + # 忽略 "cannot marshal None" 错误，因为方法返回 None 是正常的 + if "cannot marshal None" in str(e): + print("btn_generate_research 已执行（返回 None，这是正常的）") + else: + raise + +# 死循环等待 status = 'generated_research' +print("等待研究方向生成完成...") +while True: + time.sleep(5) # 每5秒检查一次 + record_data = models.execute_kw(db, uid, password, 'alpha.generate.research.direction', 'read', [[record_id], ['status', 'message']]) + if not record_data: + print("记录不存在") + exit(1) + current_status = record_data[0].get('status') + message = record_data[0].get('message', '') + print(f"当前状态: {current_status}, 消息: {message}") + + if current_status == 'generated_research': + print("研究方向生成完成！") + break + elif current_status == 'failed': + print(f"研究方向生成失败: {message}") + exit(1) + +# 4. 点击 btn_generate_prepare_feature_engineering_prompt 按钮 +print("\n开始生成特征工程提示词...") +try: + result = models.execute_kw(db, uid, password, 'alpha.generate.research.direction', 'btn_generate_prepare_feature_engineering_prompt', [[record_id]]) + print(f"btn_generate_prepare_feature_engineering_prompt 执行结果: {result}") +except Fault as e: + # 忽略 "cannot marshal None" 错误，因为方法返回 None 是正常的 + if "cannot marshal None" in str(e): + print("btn_generate_prepare_feature_engineering_prompt 已执行（返回 None，这是正常的）") + else: + raise + +# 5. 点击 btn_generate_feature 按钮 +print("\n开始生成特征工程...") +try: + result = models.execute_kw(db, uid, password, 'alpha.generate.research.direction', 'btn_generate_feature', [[record_id]]) + print(f"btn_generate_feature 执行结果: {result}") +except Fault as e: + # 忽略 "cannot marshal None" 错误，因为方法返回 None 是正常的 + if "cannot marshal None" in str(e): + print("btn_generate_feature 已执行（返回 None，这是正常的）") + else: + raise + +# 死循环等待 status = 'done' +print("等待特征工程生成完成...") +while True: + time.sleep(5) # 每5秒检查一次 + record_data = models.execute_kw(db, uid, password, 'alpha.generate.research.direction', 'read', [[record_id], ['status', 'message']]) + if not record_data: + print("记录不存在") + exit(1) + current_status = record_data[0].get('status') + message = record_data[0].get('message', '') + print(f"当前状态: {current_status}") + + if current_status == 'done': + print("特征工程生成完成！") + break + elif current_status == 'failed': + print(f"特征工程生成失败: {message}") + exit(1) + +# 6. 点击 btn_generate_idea 按钮 +print("\n开始生成 Idea...") +try: + result = models.execute_kw(db, uid, password, 'alpha.generate.research.direction', 'btn_generate_idea', [[record_id]]) + print(f"btn_generate_idea 执行结果: {result}") +except Fault as e: + # 忽略 "cannot marshal None" 错误，因为方法返回 None 是正常的 + if "cannot marshal None" in str(e): + print("btn_generate_idea 已执行（返回 None，这是正常的）") + else: + raise + +print("\n全部流程执行完成！收工！") + diff --git a/rpc_alpha_workflow/02_generate_alpha_idea.py b/rpc_alpha_workflow/02_generate_alpha_idea.py new file mode 100644 index 0000000..6cdfd15 --- /dev/null +++ b/rpc_alpha_workflow/02_generate_alpha_idea.py @@ -0,0 +1,17 @@ +# -*- coding: utf-8 -*- +from xmlrpc.client import ServerProxy + +is_local = 0 +if is_local: + url = 'http://127.0.0.1:28888' +else: + url = 'http://192.168.31.41:32000' + +username = 'rpc' +password = 'aaaAAA111' +db = "quantify" + +common = ServerProxy(f"{url}/xmlrpc/2/common") +uid = common.authenticate(db, username, password, {}) + +print(uid) \ No newline at end of file diff --git a/rpc_alpha_workflow/03_decode_template.py b/rpc_alpha_workflow/03_decode_template.py new file mode 100644 index 0000000..6cdfd15 --- /dev/null +++ b/rpc_alpha_workflow/03_decode_template.py @@ -0,0 +1,17 @@ +# -*- coding: utf-8 -*- +from xmlrpc.client import ServerProxy + +is_local = 0 +if is_local: + url = 'http://127.0.0.1:28888' +else: + url = 'http://192.168.31.41:32000' + +username = 'rpc' +password = 'aaaAAA111' +db = "quantify" + +common = ServerProxy(f"{url}/xmlrpc/2/common") +uid = common.authenticate(db, username, password, {}) + +print(uid) \ No newline at end of file diff --git a/rpc_alpha_workflow/reference_01.py b/rpc_alpha_workflow/reference_01.py new file mode 100644 index 0000000..dafece5 --- /dev/null +++ b/rpc_alpha_workflow/reference_01.py @@ -0,0 +1,519 @@ +# -*- coding: utf-8 -*- +import uuid +import sys +import httpx +from odoo import models, fields + + +class AlphaGenerateResearchDirection(models.Model): + _name = 'alpha.generate.research.direction' + _description = 'Alpha Generate Research Direction' + _order = 'id desc' + + name = fields.Char(string='Name', required=True, default=lambda self: str(uuid.uuid4())) + + status = fields.Selection([ + ('draft', 'Draft'), + ('generated_research', 'Generated Research'), + ('generated_feature', 'Generated Feature'), + ('done', 'Done'), + ('failed', 'Failed'), + ('cancel', 'Cancel') + ], string='Status', default='draft') + + region = fields.Many2one('alpha.region.settings', string='Region', required=True) + + universe = fields.Many2one('alpha.universe.settings', string='Universe', required=True) + + research_direction_prompt = fields.Many2one('alpha.prompt.settings', string='Research Direction Prompt') + + feature_engineering_prompt = fields.Many2one('alpha.prompt.settings', string='Feature Engineering Prompt') # 用于加上替换的研究方向, 组合新的工程特征提示词 + + results = fields.Text(string='Results') + + message = fields.Char(string='Message') + + llm_settings_line_id = fields.Many2one('llm.settings.line', string='Model') + + models_name = fields.Char(string='Model Name', related='llm_settings_line_id.model_name') + + provider_name = fields.Char(string='Provider Name', related='llm_settings_line_id.llm_setting_id.name') + + research_direction = fields.Text(string='Research Direction') + + prepare_feature_engineering_prompt = fields.Text(string='Feature Engineering Prompt') # 用于已组合完成, 准备生成工程特征的提示词 + + feature_engineering = fields.Text(string='Feature Engineering') + + use_datasets = fields.Many2one('alpha.datasets', string='Use Datasets') + + datasets_line_ids = fields.One2many('alpha.research.datasets.line', 'research_direction_id', string='Datasets Lines') + + def _get_callback_url(self): + """根据平台获取回调 URL""" + platform_info = sys.platform + if platform_info == "darwin": + # Mac 开发环境使用本地地址 + base_url_odoo = self.env['ir.config_parameter'].sudo().get_param('web.base.url.local', 'http://127.0.0.1:8069') + else: + base_url_odoo = self.env['ir.config_parameter'].sudo().get_param('web.base.url') + return f"{base_url_odoo}/api/alpha-generate-research-direction/result" + + def btn_generate_research(self): + """ + 生成研究方向按钮 + 通过选择的基础提示词，推送到 LLM 微服务生成研究方向 + post 无需等待，LLM 会回写结果到 research_direction 字段 + 成功后状态变为 generated_research，失败不修改状态，失败信息写到 message + """ + self.ensure_one() + + # 如果research_direction字段已存在, 则跳过不生成 + if self.research_direction: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Research direction already exists, please delete it first.', + 'sticky': False, + }, + } + + # 校验必填字段 + if not self.research_direction_prompt: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Please select a research direction prompt template.', + 'type': 'danger', + 'sticky': False + } + } + if not self.llm_settings_line_id: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Please select a model.', + 'type': 'danger', + 'sticky': False + } + } + + model_name = self.llm_settings_line_id.model_name + base_url = self.llm_settings_line_id.llm_setting_id.base_url + api_key = self.llm_settings_line_id.llm_setting_id.api_key + + final_prompt = self.research_direction_prompt.prompt + + ms_config = self.get_ms_config() + ms_url = ms_config.get('url') + + callback_url = self._get_callback_url() + + post_data = { + "record_id": self.id, + "odoo_model": self._name, + "action_type": "research_direction", + "prompt": final_prompt, + "model_name": model_name, + "base_url": base_url, + "api_key": api_key, + "callback_url": callback_url + } + + if not ms_url: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Microservice URL is not available.', + 'type': 'danger', + 'sticky': False + } + } + + # 发送请求，超短超时，不等待响应 + try: + httpx.post(f'{ms_url}:32004/api_llm_generate', json=post_data, timeout=0.001) + except httpx.TimeoutException: + pass + except Exception as e: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': f'Failed to send microservices: {e}', + 'type': 'danger', + 'sticky': False + } + } + + self.write({ + 'status': 'draft', + 'message': 'Request sent to microservice, waiting for research direction...' + }) + + def btn_generate_prepare_feature_engineering_prompt(self): + # 手动填入生成好的研究方向, 跳过第一步的 llm 生成 + self.ensure_one() + + # 如果没有研究方向, 需要生成 + if not self.research_direction or not self.feature_engineering_prompt: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Need to generate research direction and need select feature_engineering_prompt', + 'sticky': False, + }, + } + + # 将已有的研究方向, 和基础的feature_engineering_prompt组合 + # feature_engineering_prompt + research_direction + feature_engineering_prompt = self.feature_engineering_prompt.prompt + research_direction = self.research_direction + + replaced_feature_engineering_prompt = feature_engineering_prompt.replace('[#replace#]', research_direction) + + self.write({ + 'status': 'generated_research', + 'prepare_feature_engineering_prompt': replaced_feature_engineering_prompt + }) + + return True + + def btn_generate_feature(self): + """ + 生成特征工程按钮 + 通过 feature_engineering_prompt 作为提示词，推送到 LLM 微服务生成特征工程 + post 无需等待，LLM 会回写结果到 feature_engineering 字段 + 成功后状态变为 done，失败状态变为 failed，失败信息写到 message + """ + self.ensure_one() + + # 如果未生成prepare_feature_engineering_prompt, 则不允许使用此按钮 + if not self.prepare_feature_engineering_prompt: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Please generate prepare_feature_engineering_prompt first', + 'type': 'danger', + 'sticky': False + } + } + + # 如果feature_engineering字段不为空, 则提示需要清空才可以使用这个按钮, 避免重复生成 + if self.feature_engineering: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Please clear feature_engineering field first', + 'type': 'danger', + 'sticky': False + } + } + + # 校验状态 + if self.status != 'generated_research': + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Please generate research direction first.', + 'type': 'danger', + 'sticky': False + } + } + + # 校验必填字段 + if not self.research_direction: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Research direction is empty, please generate it first.', + 'type': 'danger', + 'sticky': False + } + } + if not self.llm_settings_line_id: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Please select a model.', + 'type': 'danger', + 'sticky': False + } + } + if not self.feature_engineering_prompt: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Please select a feature engineering prompt template.', + 'type': 'danger', + 'sticky': False + } + } + + model_name = self.llm_settings_line_id.model_name + base_url = self.llm_settings_line_id.llm_setting_id.base_url + api_key = self.llm_settings_line_id.llm_setting_id.api_key + + # 使用 feature_engineering_prompt 作为提示词 + final_prompt = self.feature_engineering_prompt.prompt + + ms_config = self.get_ms_config() + ms_url = ms_config.get('url') + + callback_url = self._get_callback_url() + + post_data = { + "record_id": self.id, + "odoo_model": self._name, + "action_type": "feature_engineering", + "prompt": final_prompt, + "model_name": model_name, + "base_url": base_url, + "api_key": api_key, + "callback_url": callback_url + } + + if not ms_url: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Microservice URL is not available.', + 'type': 'danger', + 'sticky': False + } + } + + # 发送请求，超短超时，不等待响应 + try: + httpx.post(f'{ms_url}:32004/api_llm_generate', json=post_data, timeout=0.001) + except httpx.TimeoutException: + pass + except Exception as e: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Failed to send microservice.', + 'type': 'danger', + 'sticky': False + } + } + + self.write({ + 'status': 'generated_research', + 'message': 'Request sent to microservice, waiting for feature engineering...' + }) + + def get_ms_config(self): + # TODO 先从 nacos 获取微服务 url + nacos_url = '' + + platform_info = sys.platform + if platform_info == "darwin": + nacos_url = 'http://192.168.31.41:30848/nacos/v1/cs/configs?dataId=microservices_dev&group=quantify' + elif platform_info.startswith("linux"): + nacos_url = 'http://192.168.31.41:30848/nacos/v1/cs/configs?dataId=microservices&group=quantify' + + try: + ms_config_resp = httpx.get(nacos_url) + ms_config_resp.raise_for_status() + except Exception as e: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'Nacos request failed.', + 'type': 'danger', + 'sticky': False + } + } + + ms_config = ms_config_resp.json() + + return ms_config + + def btn_generate_idea(self): + # 最后一步, 当工程特征已经生成, 并且已选择数据字段, 则生成idea + # 同一份特征工程, 多个数据字段, 生成与数据字段相同数量的 idea + self.ensure_one() + + # if self.status != 'generated_feature': + # return { + # 'type': 'ir.actions.client', + # 'tag': 'display_notification', + # 'params': { + # 'title': 'ERROR', + # 'message': 'Feature engineering not completed.', + # 'type': 'danger', + # 'sticky': False + # } + # } + + if not self.datasets_line_ids: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'No data fields selected.', + 'type': 'danger', + 'sticky': False + } + } + + if not self.feature_engineering: + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'ERROR', + 'message': 'No feature engineering selected.', + 'type': 'danger', + 'sticky': False + } + } + + # 开始生成 idea, 先获取到数据集的 name 和 data_sets_type + for data_field in self.datasets_line_ids: + data_fields_name = data_field.datasets_id.datasets_id + data_sets_type = '' + region_id = self.region.id + universe_id = self.universe.id + for data_field_line in data_field.datasets_id.line_ids: + data_sets_type = data_field_line.data_sets_type + break + + # 创建一个alpha.idea记录, 为了防止创建重复的 idea, 当前模型的 name 是 uuid4, 加上 region_id, universe_id, data_fields_name, data_sets_type,这几个字段, 先搜索是否存在相应的记录 + # 如果不存在, 创建, 如果存在, 跳过 + + # 根据 data_sets_type 确定 data_type + data_type = 'MATRIX' if data_sets_type == 'MATRIX' else 'VECTOR' + + # 检查是否已存在相同的 idea 记录（特征工程 + 数据集 都相同） + existing_idea = self.env['alpha.idea'].search([ + ('region', '=', region_id), + ('universe', '=', universe_id), + ('data_type', '=', data_type), + ('replace_prompt', '=', self.feature_engineering), + ('needed_data_set_ids.name', '=', data_fields_name) + ], limit=1) + + if existing_idea: + continue + + # 创建新的 alpha.idea 记录 + idea_vals = { + 'region': region_id, + 'universe': universe_id, + 'data_type': data_type, + 'delay': self.use_datasets.delay if self.use_datasets else '1', + 'replace_prompt': self.feature_engineering, + 'needed_data_set_ids': [(0, 0, {'name': data_fields_name})], + } + + # 找到最终的 meta_prompt + meta_prompt = self.env['alpha.prompt.settings'].search([('prompt_type', '=', 'meta_prompt')], limit=1) + + if meta_prompt: + idea_vals.update({'meta_prompt': meta_prompt.id}) + + new_record = self.env['alpha.idea'].create(idea_vals) + + # 创建成功之后, 顺便点一下 + try: + new_record.btn_check_and_fetch_data() + new_record.btn_generate_final_prompt() + except Exception as e: + print(f'error: {e}') + + return { + 'type': 'ir.actions.client', + 'tag': 'display_notification', + 'params': { + 'title': 'Success', + 'message': 'Ideas generated successfully.', + 'type': 'success', + 'sticky': False, + } + } + + def btn_quick_selection_datasets(self): + self.ensure_one() + + # 1, 检查是否存在已选择的数据集, 如果存在, 清空self.datasets_line_ids + if self.datasets_line_ids: + self.datasets_line_ids.unlink() + + # 2, 查询数据集, 并创建关联行 + datasets = self.env['alpha.datasets'].search([ + ('region', '=', self.region.id), + ('universe', '=', self.universe.id), + ]) + + # 3, 遍历数据, 插入到 datasets_line_ids + for dataset in datasets: + self.datasets_line_ids.create({ + 'research_direction_id': self.id, + 'datasets_id': dataset.id, + }) + + return True + + def btn_action_cancel(self): + self.ensure_one() + self.write({ + 'status': 'cancel' + }) + + +class AlphaResearchDatasetsLine(models.Model): + """研究方向数据集关联行""" + _name = 'alpha.research.datasets.line' + _description = 'Alpha Research Datasets Line' + _order = 'id desc' + + research_direction_id = fields.Many2one('alpha.generate.research.direction', string='Research Direction', required=True, ondelete='cascade') + + datasets_id = fields.Many2one('alpha.datasets', string='Dataset', required=True) + + data_field_count = fields.Integer(string='Data Field Count', related='datasets_id.data_field_count', readonly=True) + + datasets_code = fields.Char(string='Dataset Code', related='datasets_id.datasets_id', readonly=True) + + datasets_name = fields.Char(string='Dataset Name', related='datasets_id.name', readonly=True) + + region = fields.Char(string='Region', related='datasets_id.region.name', readonly=True) + + universe = fields.Char(string='Universe', related='datasets_id.universe.name', readonly=True) + + delay = fields.Selection([ + ('1', '1'), + ('0', '0') + ], string='Delay', related='datasets_id.delay', readonly=True) diff --git a/wqb-get-alphas/simple_alpha_fetcher.py b/wqb-get-alphas/simple_alpha_fetcher.py new file mode 100644 index 0000000..355b6ee --- /dev/null +++ b/wqb-get-alphas/simple_alpha_fetcher.py @@ -0,0 +1,264 @@ +import httpx +from httpx import BasicAuth, Timeout +import pandas as pd +from tqdm import tqdm +from pathlib import Path +import logging +import time +import random + + +class SimpleAlphaFetcher: + def __init__(self, base_path=None): + """ + 初始化 Alpha 获取器 + + Args: + base_path: 输出文件保存路径 + """ + self.client = None + self.base_path = Path(base_path) if base_path else Path.cwd() + self.logger = self._setup_logger() + + # 登录 + self.login() + + def _setup_logger(self): + """设置日志记录器""" + logger = logging.getLogger(__name__) + logger.setLevel(logging.INFO) + + if not logger.handlers: + handler = logging.StreamHandler() + formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') + handler.setFormatter(formatter) + logger.addHandler(handler) + + return logger + + def login(self): + """登录 WorldQuant Brain API""" + try: + # 从 nacos 获取账号密码 + with httpx.Client(timeout=10.0) as temp_client: + nacos_resp = temp_client.get( + 'http://192.168.31.41:30848/nacos/v1/cs/configs?dataId=wq_account&group=quantify' + ) + + if nacos_resp.status_code != 200: + self.logger.error('获取账号密码失败') + return False + + config = nacos_resp.json() + username = config.get('user_name') + password = config.get('password') + + if not username or not password: + self.logger.error('账号密码不完整') + return False + + self.logger.info(f"正在登录账户: {username}") + + # 创建客户端并设置超时 + timeout = Timeout(connect=30.0, read=60.0, write=30.0, pool=30.0) + self.client = httpx.Client( + auth=BasicAuth(username, password), + timeout=timeout + ) + + # 发送登录请求 + response = self.client.post('https://api.worldquantbrain.com/authentication') + + if response.status_code == 201: + self.logger.info("登录成功!") + return True + else: + self.logger.error(f"登录失败: {response.status_code} - {response.text}") + self.client.close() + self.client = None + return False + + except Exception as e: + self.logger.error(f"登录异常: {e}") + return False + + def fetch_alphas( + self, + max_pages: int = 100, + limit: int = 100, + delay: int = 1, + region: str = "USA", + universe: str = "TOP3000", + hidden: str = "false", + output_file_name: str = "alpha_list.csv", + mode: str = "w", + max_retries: int = 3, + ): + """ + 获取 Alpha 列表 + + Args: + max_pages: 最大爬取页数 + limit: 每页数量 + delay: API 请求的延迟设置 + region: 市场区域 + universe: 股票池 + hidden: 是否搜索已隐藏的 Alpha + output_file_name: 输出的 CSV 文件名 + mode: 写入模式："w"(覆盖写入) 或 "a"(追加写入) + max_retries: 请求失败时的最大重试次数 + + Returns: + list: Alpha 列表 + """ + if not self.client: + self.logger.error("客户端未登录，无法执行获取") + return [] + + fetched_alphas = [] + offset = 0 + total_accessed = 0 + pages_fetched = 0 + + # 先获取总数 + count_url = ( + f"https://api.worldquantbrain.com/users/self/alphas?stage=IS&hidden={hidden}" + f"&limit=1&settings.delay={delay}&settings.region={region}&status=UNSUBMITTED%1FIS_FAIL&settings.universe={universe}" + ) + + total_available = 0 + for attempt in range(max_retries): + try: + count_response = self.client.get(count_url) + total_available = count_response.json()["count"] + break + except Exception as e: + self.logger.warning(f"获取 Alpha 总数失败 (尝试 {attempt+1}/{max_retries}): {e}") + if attempt < max_retries - 1: + time.sleep(random.uniform(3, 5)) + else: + self.logger.error("获取 Alpha 总数最终失败") + return [] + + if total_available == 0: + self.logger.warning("未找到任何 Alpha") + return [] + + # 计算实际最大页数 + actual_max_pages = min(max_pages, (total_available + limit - 1) // limit) + self.logger.info(f"共找到 {total_available} 个 Alpha，计划爬取 {actual_max_pages} 页...") + + pbar = tqdm(total=actual_max_pages, desc="爬取 Alpha 页面", unit="页") + + while pages_fetched < actual_max_pages: + # 构建请求 URL + url = ( + f"https://api.worldquantbrain.com/users/self/alphas?stage=IS&limit={limit}" + f"&offset={offset}&settings.delay={delay}&settings.region={region}&hidden={hidden}&status=UNSUBMITTED%1FIS_FAIL&settings.universe={universe}" + ) + + try: + # 使用重试机制 + response = None + for attempt in range(max_retries): + try: + response = self.client.get(url) + break + except Exception as e: + self.logger.warning(f"请求失败 (尝试 {attempt+1}/{max_retries}): {e}") + if attempt < max_retries - 1: + time.sleep(random.uniform(3, 5)) + else: + raise + + if response.status_code == 400: + self.logger.warning(f"遇到 API 限制 (offset={offset})，停止获取") + break + + response_data = response.json() + + if not isinstance(response_data, dict) or "results" not in response_data: + self.logger.error(f"API 返回了意外的数据: {response_data}") + break + + alphas = response_data["results"] + + if not alphas: + break + + fetched_alphas.extend(alphas) + total_accessed += len(alphas) + pages_fetched += 1 + + # 更新进度条 + pbar.update(1) + pbar.set_postfix({ + "Region": region, + "已获取": total_accessed, + "本页": len(alphas), + }) + + if len(alphas) < limit: + self.logger.info(f"本页数据不足 {limit} 条，已到达末尾") + break + + offset += limit + + # 添加延迟避免请求过快 + time.sleep(0.5) + + except Exception as e: + self.logger.error(f"请求失败: {e}") + break + + pbar.close() + + if not fetched_alphas: + self.logger.warning("未获取到任何 Alpha！") + return [] + + # 按 fitness 排序 + df = pd.DataFrame(fetched_alphas) + df["temp_fitness"] = df.apply( + lambda row: row["is"].get("fitness", 0) if isinstance(row.get("is"), dict) else 0, + axis=1 + ) + df_sorted = df.sort_values(by="temp_fitness", ascending=False) + df_sorted = df_sorted.drop("temp_fitness", axis=1) + + output_path = self.base_path / output_file_name + + if mode == "w": + df_sorted.to_csv(output_path, index=False) + elif mode == "a": + df_sorted.to_csv(output_path, mode="a", index=False, header=False) + + self.logger.info(f"Alpha 列表已保存！共 {len(fetched_alphas)} 条记录！\n文件路径: {output_path}") + + return fetched_alphas + + def close(self): + """关闭客户端""" + if self.client: + self.client.close() + + +# 使用示例 +if __name__ == "__main__": + # 创建获取器实例（会自动登录） + fetcher = SimpleAlphaFetcher() + + if fetcher.client: + # 获取 Alpha 列表 + results = fetcher.fetch_alphas( + max_pages=100, # 最多爬取 100 页 + limit=100, # 每页 100 条 + region="USA", # 美股市场 + universe="TOP3000", # TOP3000 股票池 + output_file_name="alpha_list.csv" # 输出文件名 + ) + + print(f"成功获取 {len(results) if results else 0} 个 Alpha") + + # 关闭客户端 + fetcher.close()