Run the aggregation processor.
(self)
| 232 | return await task.run() |
| 233 | |
| 234 | async def run(self): |
| 235 | """ Run the aggregation processor. |
| 236 | """ |
| 237 | if self.procs_config is None: |
| 238 | raise ValueError("No processors configured for aggregation") |
| 239 | |
| 240 | info = { |
| 241 | "symbols": self.symbols, |
| 242 | "symbols_info": self.assets_info, |
| 243 | } |
| 244 | data_info = [] |
| 245 | tasks = [] |
| 246 | for proc_config in self.procs_config: |
| 247 | |
| 248 | configs = { |
| 249 | "type": proc_config.get("type", None), |
| 250 | "assets_name": proc_config.get("assets_name", None), |
| 251 | "source": proc_config.get("source", None), |
| 252 | "data_path": proc_config.get("data_path", None), |
| 253 | "data_type": proc_config.get("data_type", None), |
| 254 | "assets_path": self.assets_path, |
| 255 | "start_date": proc_config.get("start_date", None), |
| 256 | "end_date": proc_config.get("end_date", None), |
| 257 | "level": proc_config.get("level", None), |
| 258 | "format": proc_config.get("format", None), |
| 259 | "max_concurrent": self.max_concurrent, |
| 260 | "feature_type": proc_config.get("feature_type", None), |
| 261 | } |
| 262 | |
| 263 | processor = PROCESSOR.build(configs) |
| 264 | |
| 265 | tasks.append(processor) |
| 266 | |
| 267 | for i in range(0, len(tasks), self.max_concurrent): |
| 268 | batch = tasks[i:min(i + self.max_concurrent, len(tasks))] |
| 269 | batch_info = await asyncio.gather(*[self.run_task(task) for task in batch]) |
| 270 | data_info.extend(batch_info) |
| 271 | |
| 272 | info['tags'] = [item['tag'] for item in data_info] |
| 273 | info['data_info'] = {item['tag']: item for item in data_info} |
| 274 | |
| 275 | # Save the info |
| 276 | info_path = os.path.join(self.workdir, "meta_info.json") |
| 277 | with open(info_path, "w") as f: |
| 278 | json.dump(info, f, indent=4) |
| 279 | |
| 280 | logger.info(f"| Pushing processed data to Hugging Face Hub: {self.repo_id}...") |
| 281 | push_to_hub_folder( |
| 282 | hf_token=os.getenv("HF_API_KEY"), |
| 283 | endpoint=os.getenv("HF_ENDPOINT"), |
| 284 | repo_id=self.repo_id, |
| 285 | repo_type=self.repo_type, |
| 286 | folder_path=self.workdir, |
| 287 | ) |