Merge pull request #20 from C00ldudeNoonan/dagster-plus

update dagster_cloud.yaml file

Author: C00ldudeNoonan

macro_agents/dagster_cloud.yaml dagster_cloud.yamlmacro_agents/dagster_cloud.yaml renamed to dagster_cloud.yaml

@@ -2,6 +2,7 @@ locations:
   - location_name: macro_agents
     code_source:
       module_name: macro_agents.definitions
+    working_directory: ./macro_agents
     build:
-      directory: .
+      directory: ./macro_agents
 

macro_agents/pyproject.toml

@@ -31,6 +31,9 @@ dependencies = [
 dev = [
     "dagster-webserver", 
     "pytest",
+    "pytest-cov",
+    "pytest-xdist",
+    "mypy",
     "ruff",
 ]
 

macro_agents/src/macro_agents/definitions.py

@@ -7,7 +7,9 @@
 from macro_agents.defs.transformation.dbt import dbt_cli_resource
 from macro_agents.defs.agents.analysis_agent import EconomicAnalyzer
 from macro_agents.defs.agents.economic_cycle_analyzer import EconomicCycleAnalyzer
-from macro_agents.defs.agents.enhanced_economic_cycle_analyzer import EnhancedEconomicCycleAnalyzer
+from macro_agents.defs.agents.enhanced_economic_cycle_analyzer import (
+    EnhancedEconomicCycleAnalyzer,
+)
 from macro_agents.defs.agents.asset_allocation_analyzer import AssetAllocationAnalyzer
 from macro_agents.defs.agents.dspy_evaluation import FinancialEvaluator, PromptOptimizer
 from macro_agents.defs.agents.backtesting import BacktestingEngine

macro_agents/src/macro_agents/defs/agents/__init__.py

@@ -9,4 +9,3 @@
 from .economic_dashboard import *
 from .backtesting_visualization import *
 from .model_improvement_pipeline import *
-

macro_agents/src/macro_agents/defs/agents/analysis_agent.py

@@ -1,11 +1,6 @@
-import duckdb
-import polars as pl
 import dspy
 from typing import Optional, Dict, Any, List
-import io
-import json
 from datetime import datetime
-from pathlib import Path
 import dagster as dg
 from pydantic import Field
 

macro_agents/src/macro_agents/defs/agents/asset_allocation_analyzer.py

@@ -1,7 +1,5 @@
 import dspy
-import polars as pl
-from typing import Optional, Dict, Any, List, Tuple
-import json
+from typing import Optional, Dict, Any, Tuple
 from datetime import datetime
 import dagster as dg
 from pydantic import Field
@@ -12,19 +10,19 @@
 
 class AssetAllocationSignature(dspy.Signature):
     """Generate specific asset allocation recommendations based on economic cycle analysis and market trends."""
-    
+
     economic_cycle_analysis: str = dspy.InputField(
         desc="Economic cycle analysis including current cycle position and key indicators"
     )
-    
+
     market_trend_analysis: str = dspy.InputField(
         desc="Market trend analysis including sector performance and momentum indicators"
     )
-    
+
     current_portfolio_context: str = dspy.InputField(
         desc="Current portfolio context and constraints (optional)"
     )
-    
+
     allocation_recommendations: str = dspy.OutputField(
         desc="""Detailed asset allocation recommendations including:
         1. Portfolio Allocation by Asset Class:
@@ -60,32 +58,37 @@ class AssetAllocationModule(dspy.Module):
     def __init__(self):
         super().__init__()
         self.analyze_allocation = dspy.ChainOfThought(AssetAllocationSignature)
-    
-    def forward(self, economic_cycle_analysis: str, market_trend_analysis: str, current_portfolio_context: str = ""):
+
+    def forward(
+        self,
+        economic_cycle_analysis: str,
+        market_trend_analysis: str,
+        current_portfolio_context: str = "",
+    ):
         return self.analyze_allocation(
             economic_cycle_analysis=economic_cycle_analysis,
             market_trend_analysis=market_trend_analysis,
-            current_portfolio_context=current_portfolio_context
+            current_portfolio_context=current_portfolio_context,
         )
 
 
 class AssetAllocationAnalyzer(dg.ConfigurableResource):
     """Asset allocation analyzer that provides specific investment recommendations."""
-    
+
     model_name: str = Field(
         default="gpt-4-turbo-preview", description="LLM model to use for analysis"
     )
     openai_api_key: str = Field(description="OpenAI API key for DSPy")
-    
+
     def setup_for_execution(self, context) -> None:
         """Initialize DSPy when the resource is used."""
         # Initialize DSPy
         lm = dspy.LM(model=self.model_name, api_key=self.openai_api_key)
         dspy.settings.configure(lm=lm)
-        
+
         # Initialize analyzer
         self._allocation_analyzer = AssetAllocationModule()
-    
+
     @property
     def allocation_analyzer(self):
         """Get asset allocation analyzer."""
@@ -102,46 +105,48 @@ def get_latest_analysis(self, md_resource: MotherDuckResource) -> Tuple[str, str
         )
         ORDER BY analysis_type
         """
-        
+
         df = md_resource.execute_query(query, read_only=True)
-        
+
         cycle_analysis = ""
         trend_analysis = ""
-        
+
         for row in df.iter_rows(named=True):
             if row["analysis_type"] == "economic_cycle":
                 cycle_analysis = row["analysis_content"]
             elif row["analysis_type"] == "market_trends":
                 trend_analysis = row["analysis_content"]
-        
+
         return cycle_analysis, trend_analysis
 
     def generate_allocation_recommendations(
         self,
         md_resource: MotherDuckResource,
         portfolio_context: str = "",
-        context: Optional[dg.AssetExecutionContext] = None
+        context: Optional[dg.AssetExecutionContext] = None,
     ) -> Dict[str, Any]:
         """Generate asset allocation recommendations based on latest analysis."""
         if context:
             context.log.info("Retrieving latest economic and market analysis...")
-        
+
         # Get latest analysis
         cycle_analysis, trend_analysis = self.get_latest_analysis(md_resource)
-        
+
         if not cycle_analysis or not trend_analysis:
-            raise ValueError("No recent economic cycle or market trend analysis found. Please run economic_cycle_analysis first.")
-        
+            raise ValueError(
+                "No recent economic cycle or market trend analysis found. Please run economic_cycle_analysis first."
+            )
+
         if context:
             context.log.info("Generating asset allocation recommendations...")
-        
+
         # Generate recommendations
         allocation_result = self.allocation_analyzer(
             economic_cycle_analysis=cycle_analysis,
             market_trend_analysis=trend_analysis,
-            current_portfolio_context=portfolio_context
+            current_portfolio_context=portfolio_context,
         )
-        
+
         # Format results
         analysis_timestamp = datetime.now()
         result = {
@@ -151,9 +156,11 @@ def generate_allocation_recommendations(
             "model_name": self.model_name,
             "allocation_recommendations": allocation_result.allocation_recommendations,
             "portfolio_context": portfolio_context,
-            "source_analysis_timestamp": self._get_latest_analysis_timestamp(md_resource)
+            "source_analysis_timestamp": self._get_latest_analysis_timestamp(
+                md_resource
+            ),
         }
-        
+
         return result
 
     def _get_latest_analysis_timestamp(self, md_resource: MotherDuckResource) -> str:
@@ -162,14 +169,14 @@ def _get_latest_analysis_timestamp(self, md_resource: MotherDuckResource) -> str
         SELECT MAX(analysis_timestamp) as latest_timestamp
         FROM economic_cycle_analysis
         """
-        
+
         df = md_resource.execute_query(query, read_only=True)
         return df[0, "latest_timestamp"] if not df.is_empty() else ""
 
     def format_allocation_as_json(
         self,
         allocation_result: Dict[str, Any],
-        metadata: Optional[Dict[str, Any]] = None
+        metadata: Optional[Dict[str, Any]] = None,
     ) -> Dict[str, Any]:
         """Format allocation recommendations as JSON record."""
         json_result = {
@@ -180,13 +187,13 @@ def format_allocation_as_json(
             "analysis_time": allocation_result["analysis_time"],
             "model_name": allocation_result["model_name"],
             "portfolio_context": allocation_result["portfolio_context"],
-            "source_analysis_timestamp": allocation_result["source_analysis_timestamp"]
+            "source_analysis_timestamp": allocation_result["source_analysis_timestamp"],
         }
-        
+
         # Add metadata if provided
         if metadata:
             json_result.update(metadata)
-        
+
         return json_result
 
     def write_allocation_to_table(
@@ -195,24 +202,24 @@ def write_allocation_to_table(
         allocation_result: Dict[str, Any],
         output_table: str = "asset_allocation_recommendations",
         if_exists: str = "append",
-        context: Optional[dg.AssetExecutionContext] = None
+        context: Optional[dg.AssetExecutionContext] = None,
     ) -> None:
         """Write allocation recommendations to database."""
         # Format results as JSON
         json_result = self.format_allocation_as_json(
             allocation_result,
             metadata={
                 "dagster_run_id": context.run_id if context else None,
-                "dagster_asset_key": str(context.asset_key) if context else None
-            }
+                "dagster_asset_key": str(context.asset_key) if context else None,
+            },
         )
-        
+
         # Write to database
         md_resource.write_results_to_table(
             [json_result],
             output_table=output_table,
             if_exists=if_exists,
-            context=context
+            context=context,
         )
 
 
@@ -229,39 +236,39 @@ def asset_allocation_recommendations(
 ) -> Dict[str, Any]:
     """
     Asset that generates specific asset allocation recommendations based on economic analysis.
-    
+
     Returns:
         Dictionary with allocation recommendations and metadata
     """
     context.log.info("Starting asset allocation analysis...")
-    
+
     # Generate recommendations
     allocation_result = allocation_analyzer.generate_allocation_recommendations(
         md_resource=md,
         portfolio_context="General portfolio - no specific constraints",
-        context=context
+        context=context,
     )
-    
+
     # Write results to database
     context.log.info("Writing allocation recommendations to database...")
     allocation_analyzer.write_allocation_to_table(
         md_resource=md,
         allocation_result=allocation_result,
         output_table="asset_allocation_recommendations",
         if_exists="append",
-        context=context
+        context=context,
     )
-    
+
     # Return metadata
     result_metadata = {
         "analysis_completed": True,
         "analysis_timestamp": allocation_result["analysis_timestamp"],
         "model_name": allocation_result["model_name"],
         "output_table": "asset_allocation_recommendations",
         "records_written": 1,
-        "source_analysis_timestamp": allocation_result["source_analysis_timestamp"]
+        "source_analysis_timestamp": allocation_result["source_analysis_timestamp"],
     }
-    
+
     context.log.info(f"Asset allocation analysis complete: {result_metadata}")
     return result_metadata
 
@@ -279,14 +286,14 @@ def custom_asset_allocation(
 ) -> Dict[str, Any]:
     """
     Asset that generates custom asset allocation recommendations with specific portfolio context.
-    
+
     This asset can be customized with different portfolio contexts by modifying the portfolio_context parameter.
-    
+
     Returns:
         Dictionary with custom allocation recommendations and metadata
     """
     context.log.info("Starting custom asset allocation analysis...")
-    
+
     # Define custom portfolio context
     custom_context = """
     Portfolio Context:
@@ -298,24 +305,22 @@ def custom_asset_allocation(
     - Liquidity needs: Moderate (some funds may be needed within 2-3 years)
     - Tax considerations: Taxable account, prefer tax-efficient strategies
     """
-    
+
     # Generate custom recommendations
     allocation_result = allocation_analyzer.generate_allocation_recommendations(
-        md_resource=md,
-        portfolio_context=custom_context,
-        context=context
+        md_resource=md, portfolio_context=custom_context, context=context
     )
-    
+
     # Write results to database
     context.log.info("Writing custom allocation recommendations to database...")
     allocation_analyzer.write_allocation_to_table(
         md_resource=md,
         allocation_result=allocation_result,
         output_table="custom_asset_allocation_recommendations",
         if_exists="append",
-        context=context
+        context=context,
     )
-    
+
     # Return metadata
     result_metadata = {
         "analysis_completed": True,
@@ -324,8 +329,8 @@ def custom_asset_allocation(
         "output_table": "custom_asset_allocation_recommendations",
         "records_written": 1,
         "portfolio_context": "Conservative retirement-focused",
-        "source_analysis_timestamp": allocation_result["source_analysis_timestamp"]
+        "source_analysis_timestamp": allocation_result["source_analysis_timestamp"],
     }
-    
+
     context.log.info(f"Custom asset allocation analysis complete: {result_metadata}")
     return result_metadata