金融版本manus＂的核心要义,附python代码

system_prompts = '''你将扮演一位以开发复杂交易算法闻名的交易策略师。你的任务是运用编程技能，使用BackTrader Python库创建定制化的交易策略，并将其保存为Python模块。请务必在策略中记录必要信息，以便后续进行分析。你也可以编写自定义的头寸规模计算器（sizer）或指标（indicator），并保存为独立模块，从而构建更复杂的策略。完成策略开发后，你可以使用提供的工具进行回测，评估其表现并做出必要调整。编码过程中创建的所有文件将自动存放在 {work_dir} 目录中，无需额外指定路径前缀。但在调用回测函数时，模块路径需采用 {work_dir.strip('/')}.<模块路径> 格式，且保存图表（savefig）的路径也需考虑 {work_dir} 目录。当策略准备就绪可进行测试时，请回复EXIT指令给执行器。'''system_prompts_en = '''You are a trading strategist known for your expertise in developing sophisticated trading algorithms.         Your task is to leverage your coding skills to create a customized trading strategy using the BackTrader Python library, and save it as a Python module.         Remember to log necessary information in the strategy so that further analysis could be done.        You can also write custom sizer / indicator and save them as modules, which would allow you to generate more sophisticated strategies.        After creating the strategy, you may backtest it with the tool you're provided to evaluate its performance and make any necessary adjustments.        All files you created during coding will automatically be in `{work_dir}`, no need to specify the prefix.         But when calling the backtest function, module path should be like `{work_dir.strip('/')}.<module_path>` and savefig path should consider `{work_dir}` as well.        Reply EXIT to executer when the strategy is ready to be tested.'''

import osimport jsonimport importlibimport pandas as pdimport backtrader as btfrom typing import Annotated, List, Tuplefrom agentscope.message import TextBlockfrom agentscope.tool import ToolResponsefrom matplotlib import pyplot as pltfrom pprint import pformatfrom core.datafeed.sqlite_dataloader import SQLiteDataLoaderclass DeployedCapitalAnalyzer(bt.Analyzer):    def start(self):        self.deployed_capital = []        self.initial_cash = self.strategy.broker.get_cash()  # Initial cash in account    def notify_order(self, order):        if order.status in [order.Completed]:            if order.isbuy():                self.deployed_capital.append(order.executed.price * order.executed.size)            elif order.issell():                self.deployed_capital.append(order.executed.price * order.executed.size)    def stop(self):        total_deployed = sum(self.deployed_capital)        final_cash = self.strategy.broker.get_value()        net_profit = final_cash - self.initial_cash        if total_deployed > 0:            self.retn = net_profit / total_deployed        else:            self.retn = 0    def get_analysis(self):        return {"return_on_deployed_capital": self.retn}class BackTraderUtils:    @staticmethod    def back_test(        ticker_symbol: Annotated[            str, "Ticker symbol of the stock (e.g., 'AAPL' for Apple)"        ],        start_date: Annotated[            str, "Start date of the historical data in 'YYYY-MM-DD' format"        ],        end_date: Annotated[            str, "End date of the historical data in 'YYYY-MM-DD' format"        ],        strategy: Annotated[            str,            "BackTrader Strategy class to be backtested. Can be pre-defined or custom. Pre-defined options: 'SMA_CrossOver'. If custom, provide module path and class name as a string like 'my_module:TestStrategy'.",        ],        strategy_params: Annotated[            str,            "Additional parameters to be passed to the strategy class formatted as json string. E.g. {'fast': 10, 'slow': 30} for SMACross.",        ] = "",        sizer: Annotated[            int | str | None,            "Sizer used for backtesting. Can be a fixed number or a custom Sizer class. If input is integer, a corresponding fixed sizer will be applied. If custom, provide module path and class name as a string like 'my_module:TestSizer'.",        ] = None,        sizer_params: Annotated[            str,            "Additional parameters to be passed to the sizer class formatted as json string.",        ] = "",        indicator: Annotated[            str | None,            "Custom indicator class added to strategy. Provide module path and class name as a string like 'my_module:TestIndicator'.",        ] = None,        indicator_params: Annotated[            str,            "Additional parameters to be passed to the indicator class formatted as json string.",        ] = "",        cash: Annotated[            float, "Initial cash amount for the backtest. Default to 10000.0"        ] = 10000.0,        save_fig: Annotated[            str | None, "Path to save the plot of backtest results. Default to None."        ] = None,    ) -> str:        """        Use the Backtrader library to backtest a trading strategy on historical stock data.        args:            ticker_symbol: Ticker symbol of the stock (e.g., 'AAPL' for Apple)            start_date: Start date of the historical data in 'YYYY-MM-DD' format            end_date: End date of the historical data in 'YYYY-MM-DD' format            strategy: BackTrader Strategy class to be backtested. Can be pre-defined or custom. Pre-defined options: 'SMA_CrossOver'. If custom, provide module path and class name as a string like 'my_module:TestStrategy'.            strategy_params: Additional parameters to be passed to the strategy class formatted as json string. E.g. {'fast': 10, 'slow': 30} for SMACross.            sizer: Sizer used for backtesting. Can be a fixed number or a custom Sizer class. If input is integer, a corresponding fixed sizer will be applied. If custom, provide module path and class name as a string like 'my_module:TestSizer'.            sizer_params: Additional parameters to be passed to the sizer class formatted as json string.        """        cerebro = bt.Cerebro()        assert (                ":" in strategy        ), "Custom strategy should be module path and class name separated by a colon."        module_path, class_name = strategy.split(":")        print(strategy)        import sys        print(sys.path)        module = importlib.import_module(module_path)        strategy_class = getattr(module, class_name)        strategy_params = json.loads(strategy_params) if strategy_params else {}        cerebro.addstrategy(strategy_class, **strategy_params)        loader = SQLiteDataLoader()        df = loader.read_df(symbols=[ticker_symbol])        df.set_index('date', inplace=True)        df.index = pd.to_datetime(df.index)        print(df)        # Create a data feed        data = bt.feeds.PandasData(            dataname=df#yf.download(ticker_symbol, start_date, end_date, auto_adjust=True)        )        cerebro.adddata(data)  # Add the data feed        # Set our desired cash start        cerebro.broker.setcash(cash)        # Set the size of the trades        if sizer is not None:            if isinstance(sizer, int):                cerebro.addsizer(bt.sizers.FixedSize, stake=sizer)            else:                assert (                    ":" in sizer                ), "Custom sizer should be module path and class name separated by a colon."                module_path, class_name = sizer.split(":")                module = importlib.import_module(module_path)                sizer_class = getattr(module, class_name)                sizer_params = json.loads(sizer_params) if sizer_params else {}                cerebro.addsizer(sizer_class, **sizer_params)        # Set additional indicator        if indicator is not None:            assert (                ":" in indicator            ), "Custom indicator should be module path and class name separated by a colon."            module_path, class_name = indicator.split(":")            module = importlib.import_module(module_path)            indicator_class = getattr(module, class_name)            indicator_params = json.loads(indicator_params) if indicator_params else {}            cerebro.addindicator(indicator_class, **indicator_params)        # Attach analyzers        cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe_ratio")        cerebro.addanalyzer(bt.analyzers.DrawDown, _name="draw_down")        cerebro.addanalyzer(bt.analyzers.Returns, _name="returns")        cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trade_analyzer")        # cerebro.addanalyzer(DeployedCapitalAnalyzer, _name="deployed_capital")        stats_dict = {"Starting Portfolio Value:": cerebro.broker.getvalue()}        results = cerebro.run()  # run it all        first_strategy = results[0]        # Access analysis results        stats_dict["Final Portfolio Value"] = cerebro.broker.getvalue()        # stats_dict["Deployed Capital"] = pformat(        #     first_strategy.analyzers.deployed_capital.get_analysis(), indent=4        # )        stats_dict["Sharpe Ratio"] = (            first_strategy.analyzers.sharpe_ratio.get_analysis()        )        stats_dict["Drawdown"] = first_strategy.analyzers.draw_down.get_analysis()        stats_dict["Returns"] = first_strategy.analyzers.returns.get_analysis()        stats_dict["Trade Analysis"] = (            first_strategy.analyzers.trade_analyzer.get_analysis()        )        if save_fig:            directory = os.path.dirname(save_fig)            if directory:                os.makedirs(directory, exist_ok=True)            plt.figure(figsize=(12, 8))            cerebro.plot()            plt.savefig(save_fig)            plt.close()        #return "Back Test Finished. Results: \n" + pformat(stats_dict, indent=2)        return ToolResponse(            content=[                TextBlock(                    type="text",                    text="Back Test Finished. Results: \n" + pformat(stats_dict, indent=2)                ),            ],        )if __name__ == "__main__":    # Example usage:    start_date = "2011-01-01"    end_date = "2012-12-31"    ticker = "159915.SZ"    # BackTraderUtils.back_test(    #     ticker, start_date, end_date, "SMA_CrossOver", {"fast": 10, "slow": 30}    # )    BackTraderUtils.back_test(        ticker,        start_date,        end_date,        "test_module:TestStrategy",        '{"exitbars": 5}',    )

async def creating_react_agent() -> None:    """创建一个 ReAct 智能体并运行一个简单任务。"""    # 准备工具    toolkit = Toolkit()    toolkit.register_tool_function(BackTraderUtils.back_test)    toolkit.register_tool_function(CodingUtils.list_dir) # 列出目录下的文件列表    toolkit.register_tool_function(CodingUtils.see_file) # 读文件内容    toolkit.register_tool_function(CodingUtils.create_file_with_code) # 代码写到文件里    toolkit.register_tool_function(CodingUtils.modify_code) #替换部分代码