wqb/AlphaGenerator
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
toor
2025-11-24 22:35:47 +08:00
parent 3508a1b496
commit f2a149d2cf
10 changed files with 269 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+5
View File
@@ -63,3 +63,8 @@ target/
other/split_clash_config/split_config other/split_clash_config/split_config
ai_news/save_data ai_news/save_data
daily/*.txt daily/*.txt
result/
./alpha.txt
./account.txt
ai_result/20251124170421.txt
+5
View File
@@ -0,0 +1,5 @@
你好！😊 很高兴见到你！
看起来你在问候后面加了一些空格，不管怎样，我都很开心能和你聊天！我是DeepSeek，一个热心的AI助手，随时准备为你提供帮助。
无论你想聊什么话题，需要解答问题，还是想要一些建议，我都会尽我所能地帮助你。今天有什么特别想了解或讨论的吗？✨
ai_result/20251124172559.txt
+5
View File
@@ -0,0 +1,5 @@
rank(((ts_rank(close,10)-ts_rank(close,50)))*rank(ts_rank(volume,5)))
rank(ts_corr(ts_rank(close,20),ts_rank(volume,5),10))
-1*rank((close-ts_mean(close,10))/ts_std_dev(close,10)*ts_mean(volume,30))
rank(ts_delta(close,15)*rank(ts_corr((high+low)/2,volume,15)))
-1*rank(ts_rank(close,5)-ts_rank(close,60))*rank(ts_std_dev(close,20)*ts_mean(volume,40))
ai_result/20251124174559.txt
+5
View File
@@ -0,0 +1,5 @@
ts_mean(group_rank(ts_mean(return_60(close), 20)), 5) / ts_std_dev(return_60(close), 20)
group_rank(ts_rank(ts_mean(close/vwap, 60), 20)) - group_rank(ts_rank(close/vwap, 5))
ts_delay(group_rank(ts_corr(ts_mean(close, 5), ts_mean(volume, 5), 10)), 5) * ts_mean(return_20(close), 5)
group_rank(ts_mean(return_20(close), 10)) / group_rank(ts_std_dev(return_5(close), 20))
group_rank(ts_mean(return_10(close), 5)) - group_mean(ts_mean(return_10(close), 5))
ai_result/20251124175745.txt
+20
View File
@@ -0,0 +1,20 @@
Here are the five industry factors using WebSim functions:
```python
group_mean(ts_return(close, 252), industry) # Industry Momentum Factor
group_mean((close / ts_mean(close, 252) - 1), industry) # Valuation Repair Factor
group_mean(ts_sum((close - ts_lag(close, 1)) * volume, 21), industry) # Capital Flow Factor
group_mean(ts_std(ts_return(close, 1), 21), industry) # Volatility Adjustment Factor
group_mean(ts_rank(close, 252), industry) # Relative Strength Factor
```
These factors:
1. Measure industry momentum using 12-month returns
2. Evaluate valuation through 200-day mean reversion
3. Track capital flow with 1-month price-volume accumulation
4. Assess volatility risk with 21-day standard deviation
5. Gauge cross-sectional strength via 12-month price ranking
Each is calculated by averaging stock-level metrics within industry groups using `group_mean()` for industry-specific signals.
ai_result/20251124180630.txt
+18
View File
@@ -0,0 +1,18 @@
```python
# 行业动量因子（相对强度）
group_rank(ts_mean(close / close[3], 3), 'industry')
# 行业估值修复因子（均值回归）
group_rank(ts_mean(close, 3) / close, 'industry')
# 行业资金流向因子（量价结合）
group_rank(ts_sum(volume * (close - close[1]), 3), 'industry')
# 行业波动率特征因子（风险调整）
group_rank(1 / (ts_std(close / close[1], 3) + 1e-8), 'industry')
# 行业间相对强度因子（横截面）
group_rank(close / ts_max(close, 3), 'industry')
```
config.json
+7
View File
@@ -0,0 +1,7 @@
{
"siliconflow": {
"base_url": "https://api.siliconflow.cn/v1",
"api_keys": "sk-pvdiisdowmuwkrpnxsrlhxaovicqibmlljwrwwvbbdjaitdl",
"model": "Qwen/Qwen3-VL-235B-A22B-Instruct"
}
}
main.py
+130 -217
View File
@@ -1,220 +1,133 @@
"""Alpha 因子表达式生成器 - 增强随机版本""" import os
import json
import random import openai
import itertools from datetime import datetime
from typing import List
class AlphaGenerator: def load_config(config_file="config.json"):
"""增强版因子表达式生成器，支持随机生成多样化因子""" try:
with open(config_file, 'r', encoding='utf-8') as f:
def __init__(self, seed=None): return json.load(f)
if seed: except FileNotFoundError:
random.seed(seed) print(f"创建配置文件: {config_file}")
with open(config_file, 'w', encoding='utf-8') as f:
# 基础操作符和函数库 json.dump({}, f, indent=4)
OPERATORS = ['+', '-', '*', '/'] print("请在config.json中配置siliconflow的api信息")
UNARY_OPERATORS = ['-', 'abs', 'rank', 'log', 'power', 'sqrt'] return None
TIME_SERIES_FUNCTIONS = [
'delay', 'ts_mean', 'ts_std_dev', 'ts_rank', 'ts_decay_linear',
'ts_corr', 'ts_skewness', 'ts_kurtosis', 'ts_max', 'ts_min', def create_prompt_file():
'ts_sum', 'ts_product', 'ts_delta', 'ts_returns' if not os.path.exists("prompt.txt"):
] with open("prompt.txt", 'w', encoding='utf-8') as f:
GROUP_FUNCTIONS = [ f.write("")
'group_rank', 'group_mean', 'group_std', 'group_neutralize', print("已创建prompt.txt，请填入提示词后重新运行")
'group_zscore', 'group_percentile' return False
] return True
CONDITIONAL_FUNCTIONS = [
'if_else', 'trade_when', 'condition_filter', 'signal_when'
] def read_prompt():
with open("prompt.txt", 'r', encoding='utf-8') as f:
# 基础字段模板 prompt = f.read().strip()
BASE_FIELDS = ['open', 'close', 'high', 'low', 'volume', 'returns', if not prompt:
'vwap', 'turnover', 'volatility', 'cap', 'market_cap'] print("prompt.txt是空的，请填入提示词")
return None
def generate_random_expression(self, datafields, depth=0, max_depth=3): return prompt
"""递归生成随机表达式"""
if depth >= max_depth or random.random() < 0.3:
# 叶子节点 - 选择基础字段或常数 def create_result_folder():
if random.random() < 0.7 and datafields: if not os.path.exists("ai_result"):
field = random.choice(datafields) os.makedirs("ai_result")
return field return "ai_result"
else:
return f"{random.uniform(0.1, 10.0):.2f}"
def call_siliconflow(api_key, prompt, model, base_url):
# 选择操作类型 try:
op_type = random.choice(['binary', 'unary', 'ts_function', 'group_function', 'conditional']) client = openai.OpenAI(
api_key=api_key,
if op_type == 'binary': base_url=base_url
left = self.generate_random_expression(datafields, depth + 1, max_depth) )
right = self.generate_random_expression(datafields, depth + 1, max_depth)
op = random.choice(self.OPERATORS) response = client.chat.completions.create(
return f"({left} {op} {right})" model=model,
messages=[{"role": "user", "content": prompt}]
elif op_type == 'unary': )
arg = self.generate_random_expression(datafields, depth + 1, max_depth)
op = random.choice(self.UNARY_OPERATORS) return response.choices[0].message.content
if op in ['-', 'abs']:
return f"{op}({arg})" except openai.AuthenticationError:
else: print("API密钥错误")
return f"{op}({arg}, {random.randint(1, 30)})" except openai.RateLimitError:
print("调用频率限制")
elif op_type == 'ts_function': except openai.APIError as e:
func = random.choice(self.TIME_SERIES_FUNCTIONS) print(f"API错误: {e}")
arg1 = self.generate_random_expression(datafields, depth + 1, max_depth) except Exception as e:
lookback = random.choice([5, 10, 20, 30, 60, 90, 252]) print(f"其他错误: {e}")
return None
if func in ['ts_corr', 'ts_covariance']:
arg2 = self.generate_random_expression(datafields, depth + 1, max_depth)
return f"{func}({arg1}, {arg2}, {lookback})" def save_result(result, folder):
else: timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
return f"{func}({arg1}, {lookback})" filename = f"{timestamp}.txt"
filepath = os.path.join(folder, filename)
elif op_type == 'group_function':
func = random.choice(self.GROUP_FUNCTIONS) with open(filepath, 'w', encoding='utf-8') as f:
arg = self.generate_random_expression(datafields, depth + 1, max_depth) f.write(result)
group_by = random.choice(['subindustry', 'industry', 'sector', 'bucket(cap, range="0.1,1,0.1")']) print(f"结果保存到: {filepath}")
return f"{func}({arg}, {group_by})"
else: # conditional def get_ai_config(config):
func = random.choice(self.CONDITIONAL_FUNCTIONS) if not config:
condition = self.generate_random_expression(datafields, depth + 1, max_depth) return None
true_val = self.generate_random_expression(datafields, depth + 1, max_depth)
false_val = self.generate_random_expression(datafields, depth + 1, max_depth) service_name = list(config.keys())[0]
return f"{func}({condition} > 0, {true_val}, {false_val})" service_config = config[service_name]
def generate_basic_strategies(self, datafields, count=50): required_fields = ["api_keys", "model", "base_url"]
"""生成基础单因子策略""" for field in required_fields:
strategies = [] if field not in service_config:
print(f"缺少配置字段: {field}")
for _ in range(count): return None
# 随机选择1-3个字段组合
num_fields = random.randint(1, 3) return service_name, service_config
selected_fields = random.sample(datafields, min(num_fields, len(datafields)))
# 生成随机表达式 def main():
expr = self.generate_random_expression(selected_fields) config = load_config()
if not config:
# 添加分组操作 return
if random.random() < 0.7:
group_by = random.choice(['subindustry', 'industry']) if not create_prompt_file():
expr = f"group_rank({expr}, {group_by})" return
strategies.append(expr) prompt = read_prompt()
if not prompt:
return strategies return
def generate_multi_factor_strategies(self, datafields, count=50): config_result = get_ai_config(config)
"""生成多因子组合策略""" if not config_result:
strategies = [] return
for _ in range(count): service_name, service_config = config_result
# 选择更多字段进行复杂组合
num_fields = random.randint(2, 5) print(f"使用服务: {service_name}")
selected_fields = random.sample(datafields, min(num_fields, len(datafields))) print(f"使用模型: {service_config['model']}")
# 生成更复杂的表达式 folder = create_result_folder()
expr = self.generate_random_expression(selected_fields, max_depth=4)
print("正在调用AI...")
# 添加中性化或标准化 result = call_siliconflow(
if random.random() < 0.6: service_config["api_keys"],
neutral_type = random.choice(['group_neutralize', 'regression_neut', 'vector_neut']) prompt,
if neutral_type == 'group_neutralize': service_config["model"],
group_by = random.choice(['subindustry', 'industry', 'sector']) service_config["base_url"]
expr = f"{neutral_type}({expr}, {group_by})" )
else:
# 为回归中性化选择另一个因子 if result:
other_field = random.choice([f for f in datafields if f not in selected_fields] or selected_fields) print(f"AI回复: {result[:200]}...")
expr = f"{neutral_type}({expr}, {other_field})" save_result(result, folder)
else:
strategies.append(expr) print("AI调用失败")
return strategies
def generate_advanced_strategies(self, datafields, count=30):
"""生成高级策略（均值回归、动量、波动率策略等）"""
strategies = []
# 均值回归策略
for _ in range(count // 3):
field = random.choice(datafields)
lookback = random.choice([5, 10, 20, 30])
expr = f"-ts_rank({field} - ts_mean({field}, {lookback}), 20)"
strategies.append(f"group_neutralize({expr}, subindustry)")
# 动量策略
for _ in range(count // 3):
field = random.choice(datafields)
short_lookback = random.choice([1, 2, 3, 5])
long_lookback = random.choice([10, 20, 30, 60])
expr = f"ts_mean({field}, {short_lookback}) - ts_mean({field}, {long_lookback})"
strategies.append(f"group_rank({expr}, subindustry)")
# 波动率策略
for _ in range(count // 3):
field = random.choice(datafields)
vol_lookback = random.choice([10, 20, 30])
expr = f"{field} / ts_std_dev({field}, {vol_lookback})"
strategies.append(f"group_rank({expr}, subindustry)")
return strategies
def generate_strategies(self, datafields, mode=1, count=100):
"""
生成因子表达式列表
参数:
datafields: 数据字段列表
mode: 生成模式 (1=基础策略, 2=多因子组合, 3=高级策略, 4=混合模式)
count: 生成数量
返回:
因子表达式列表
"""
all_fields = datafields + self.BASE_FIELDS
unique_fields = list(set(all_fields)) # 去重
if mode == 1:
return self.generate_basic_strategies(unique_fields, count)
elif mode == 2:
return self.generate_multi_factor_strategies(unique_fields, count)
elif mode == 3:
return self.generate_advanced_strategies(unique_fields, count)
elif mode == 4:
# 混合模式 - 从所有类型中随机选择
basic = self.generate_basic_strategies(unique_fields, count // 3)
multi = self.generate_multi_factor_strategies(unique_fields, count // 3)
advanced = self.generate_advanced_strategies(unique_fields, count // 3)
return basic + multi + advanced
else:
print("❌ 无效的生成模式，使用混合模式")
return self.generate_strategies(unique_fields, mode=4, count=count)
# 使用示例
if __name__ == "__main__": if __name__ == "__main__":
# 示例数据字段 main()
sample_fields = ['volume', 'close', 'open', 'high', 'low', 'returns', 'turnover', 'pe_ratio', 'pb_ratio']
generator = AlphaGenerator(seed=42) # 设置随机种子以便复现
print("=== 基础策略因子 (20个示例) ===")
basic_strategies = generator.generate_strategies(sample_fields, mode=1, count=20)
for i, strategy in enumerate(basic_strategies[:10], 1): # 只显示前10个
print(f"{i}. {strategy}")
print(f"\n共生成 {len(basic_strategies)} 个基础因子")
print("\n=== 多因子组合策略 (20个示例) ===")
multi_strategies = generator.generate_strategies(sample_fields, mode=2, count=20)
for i, strategy in enumerate(multi_strategies[:10], 1):
print(f"{i}. {strategy}")
print(f"\n共生成 {len(multi_strategies)} 个多因子组合")
print("\n=== 随机测试 - 每次运行生成不同的因子 ===")
for run in range(3):
print(f"\n--- 第{run+1}次运行 ---")
random_strategies = generator.generate_strategies(sample_fields, mode=4, count=5)
for i, strategy in enumerate(random_strategies, 1):
print(f"{i}. {strategy}")
prompt.txt
+37
View File
@@ -0,0 +1,37 @@
【重要平台约束】
平台：WorldQuant WebSim
可用函数：ts_*(时间序列), group_*(横截面), 基础数学运算
可用字段：open, high, low, close, volume, vwap, sector, industry, country, market_cap
禁用函数：pandas, numpy, 自定义Python函数，机器学习库
输出格式：严格一行WebSim表达式，使用WQ平台函数
表达式要求：必须能直接在WebSim中运行
我正在WorldQuant研究行业轮动策略，请设计行业选择因子：
**策略背景**：
- 目标：在不同经济周期中选择强势行业
- 数据：使用sector, industry分类字段
- 频率：月度调仓
- 风险：控制行业集中度
**需要5个行业因子**：
1. 行业动量因子（相对强度）
2. 行业估值修复因子（均值回归）
3. 行业资金流向因子（量价结合）
4. 行业波动率特征因子（风险调整）
5. 行业间相对强度因子（横截面）
**具体要求**：
- 必须使用group函数处理行业分组
- 每个因子包含清晰的行业轮动逻辑
- 考虑行业特性的持久性
- 提供经济周期适应的解释
**输出格式**(一行一个, 不要输出多余的东西)：
[表达式]
[表达式]
[表达式]
...
请提供具体的WQ表达式。
重申：请确保所有表达式都使用WorldQuant WebSim平台函数，不要使用pandas、numpy或其他Python库函数。输出必须是一行有效的WQ表达式。
siliconflow_detail.py
+37
View File
@@ -0,0 +1,37 @@
# -*- coding: utf-8 -*-
import json
import httpx
with open('config.json', 'r') as f:
config = json.load(f)
siliconflow = config['siliconflow']
token = siliconflow['api_keys']
headers = {"Authorization": f"Bearer {token}"}
def get_user_info():
# 获取用户信息
url = "https://api.siliconflow.cn/v1/user/info"
response = httpx.get(url, headers=headers)
for key, value in response.json().items():
if key != "data":
print(f"{key}: {value}")
else:
for sub_key, sub_value in value.items():
print(f"{sub_key}: {sub_value}")
def get_models_info():
url = "https://api.siliconflow.cn/v1/models"
response = httpx.get(url, headers=headers)
data = response.json()['data']
for d in data:
print(d)
# get_models_info()
# get_user_info()