全天候风险平价策略下载 | Quantlab AI v0.2：OpenAI的Swarm适配国内大模型（附python代码下载）

原创内容第679篇，专注量化投资、个人成长与财富自由。

今天我们来实现服务端策略下载，下载后支持在本地调试运作，及查看源代码。

通过服务器下载策略的代码：

@login_required
def down_strategy(request, task_id: str):task = mongo_utils.get_db()['tasks'].find_one({'task_id': task_id})if not task:return {'message': f'{task_id}不存在。'}else:del task['update_datetime']json_string = json.dumps(task)# 创建HTTP响应并设置内容类型为'text/plain'response = HttpResponse(json_string, content_type='text/plain')# 设置文件名response['Content-Disposition'] = 'attachment; filename="{}.json"'.format(task_id)return response

点击“下载”按钮，可以把策略下载到本地：

把下载的策略文件置于data/tasks目录下：

运行后可以看到效果：

后续策略均通过这个形式发布。

代码下载：AI量化实验室——2024量化投资的星辰大海

AI量化实验室星球，已经运行三年多，1200+会员。

quantlab代码交付至5.X版本，含几十个策略源代码，因子表达式引擎、遗传算法（Deap）因子挖掘引擎等，每周五迭代一次。

（国庆优惠券）

quantlab_ai v0.2

续前文：quantlab_ai版本v0.1代码发布：从研报中提取因子并建模（附代码与研报集下载）

openapi最新开源的swarm多智能体框架。

https://github.com/openai/swarm

代码量不大，不需要通过pip安装，直接把代码下载回来，整合到quantlab版本中。

一共就四个核心文件。

网上的基本介绍：

OpenAI开源发布了一项重量级研究成果：Swarm。这是一个实验性质的多智能体编排框架，主打特征是工效（ergonomic）与轻量（lightweight）。客观讲，我非常看重lightweight。

我们不必要在框架本身浪费太多时间。

它提供了两个基本的抽象概念：智能体（Agent）和交接（handoffs）。通过它们来实现智能体之间的协作和执行。

老规矩，先跑通hello world。

默认需要使用gpt-4o。我把它替换成KIMI的API。

from swarm import Swarm, Agent
from core.llm import LLMllm = LLM().client
client = Swarm(client=llm)def transfer_to_agent_b():return agent_bagent_a = Agent(name="Agent A",instructions="You are a helpful agent.",functions=[transfer_to_agent_b],
)agent_b = Agent(name="Agent B",instructions="Only speak in Haikus.",
)response = client.run(agent=agent_a,messages=[{"role": "user", "content": "I want to talk to agent B."}],model_override='moonshot-v1-128k'
)print(response.messages[-1]["content"])

对swarm的代码进行改造，已适配国内的API。

class Swarm:def __init__(self, api_key=os.getenv('KIMI_KEY'), base_url="https://api.moonshot.cn/v1", model='moonshot-v1-128k',client=None):if not client:client = OpenAI(api_key=api_key,base_url=base_url,)self.client = clientself.model_name = model

在run里加入模型：

def run(self,agent: Agent,messages: List,context_variables: dict = {},model_override: str = None,stream: bool = False,debug: bool = False,max_turns: int = float("inf"),execute_tools: bool = True,
) -> Response:if self.model_name:model_override = self.model_nameif stream:return self.run_and_stream(agent=agent,messages=messages,context_variables=context_variables,model_override=model_override,debug=debug,max_turns=max_turns,execute_tools=execute_tools,)

看一下真实的例子——天气预报：

其实就是fuction call，但非常易用，定义两个python函数，一个是获取天气，一个是发送email。

import jsonfrom swarm import Agentdef get_weather(location, time="now"):"""Get the current weather in a given location. Location MUST be a city."""return json.dumps({"location": location, "temperature": "65", "time": time})def send_email(recipient, subject, body):print("Sending email...")print(f"To: {recipient}")print(f"Subject: {subject}")print(f"Body: {body}")return "Sent!"weather_agent = Agent(name="Weather Agent",instructions="You are a helpful agent.",functions=[get_weather, send_email],
)

会自动识别需求后，自动调用函数