加载factors文件夹中的数据
(trigger_time: str)
| 761 | |
| 762 | |
| 763 | def load_factors_data(trigger_time: str) -> Dict: |
| 764 | """加载factors文件夹中的数据""" |
| 765 | from contest_trade.config.config import PROJECT_ROOT |
| 766 | |
| 767 | factors_data = { |
| 768 | 'trigger_time': trigger_time, |
| 769 | 'agents': {} |
| 770 | } |
| 771 | |
| 772 | # 格式化时间戳用于文件匹配 |
| 773 | if trigger_time and trigger_time != 'N/A': |
| 774 | timestamp_str = trigger_time.replace("-", "-").replace(":", "-").replace(" ", "_") |
| 775 | else: |
| 776 | return factors_data |
| 777 | |
| 778 | # 读取factors目录 |
| 779 | factors_dir = Path(PROJECT_ROOT) / "agents_workspace" / "factors" |
| 780 | if not factors_dir.exists(): |
| 781 | return factors_data |
| 782 | |
| 783 | try: |
| 784 | for agent_dir in factors_dir.iterdir(): |
| 785 | if agent_dir.is_dir(): |
| 786 | agent_name = agent_dir.name |
| 787 | |
| 788 | # 查找对应时间戳的JSON文件 |
| 789 | pattern = f"{timestamp_str}*.json" |
| 790 | files = list(agent_dir.glob(pattern)) |
| 791 | |
| 792 | if files: |
| 793 | # 读取第一个匹配的文件 |
| 794 | with open(files[0], 'r', encoding='utf-8') as f: |
| 795 | agent_data = json.load(f) |
| 796 | factors_data['agents'][agent_name] = agent_data |
| 797 | except Exception as e: |
| 798 | console.print(f"[yellow]加载factors数据时出错: {e}[/yellow]") |
| 799 | |
| 800 | return factors_data |
| 801 | |
| 802 | |
| 803 | def display_detailed_report(final_state: Dict): |
no outgoing calls
no test coverage detected