A Coding Guide to Build an Autonomous Agentic AI for Time Series Forecasting with Darts and Hugging Face

!pip set up darts transformers pandas matplotlib numpy -q


import pandas as pd
import numpy as np
from darts import TimeSeries
from darts.fashions import ExponentialSmoothing, NaiveSeasonal, LinearRegressionMannequin
from darts.metrics import mape, rmse
from transformers import pipeline
import matplotlib.pyplot as plt
from datetime import datetime, timedelta

class TimeSeriesAgent:
   """Autonomous agent for time collection evaluation and forecasting"""
  
   def __init__(self):
       print(" Initializing Agent Brain...")
       self.llm = pipeline("text-generation", mannequin="distilgpt2", max_length=150,
                          do_sample=True, temperature=0.7)
      
       self.fashions = {
           'exponential_smoothing': ExponentialSmoothing(),
           'naive_seasonal': NaiveSeasonal(Okay=12),
           'linear_regression': LinearRegressionMannequin(lags=12)
       }
       self.selected_model = None
       self.forecast = None
      
   def understand(self, knowledge):
       """Agent perceives and analyzes the time collection knowledge"""
       print("n  PERCEPTION PHASE")
       self.ts = TimeSeries.from_dataframe(knowledge, 'date', 'worth', freq='M')
      
       development = "growing" if knowledge['value'].iloc[-1] > knowledge['value'].iloc[0] else "reducing"
       volatility = knowledge['value'].std() / knowledge['value'].imply()
       seasonality = self._detect_seasonality(knowledge['value'])
      
       evaluation = {
           'size': len(knowledge),
           'development': development,
           'volatility': f"{volatility:.2f}",
           'has_seasonality': seasonality,
           'imply': f"{knowledge['value'].imply():.2f}",
           'vary': f"{knowledge['value'].min():.2f} to {knowledge['value'].max():.2f}"
       }
      
       print(f" Data Points: {evaluation['length']}")
       print(f" Trend: {evaluation['trend'].higher()}")
       print(f" Volatility: {evaluation['volatility']}")
       print(f" Seasonality: {'Detected' if seasonality else 'Not detected'}")
      
       return evaluation
  
   def _detect_seasonality(self, collection, threshold=0.3):
       """Simple seasonality detection"""
       if len(collection) < 24:
           return False
       acf = np.correlate(collection - collection.imply(), collection - collection.imply(), mode='full')
       acf = acf[len(acf)//2:]
       acf /= acf[0]
       return np.max(acf[12:24]) > threshold if len(acf) > 24 else False
  
   def purpose(self, evaluation):
       """Agent causes about which mannequin to use"""
       print("n REASONING PHASE")
      
       immediate = f"Time collection evaluation: {evaluation['length']} knowledge factors, {evaluation['trend']} development, " 
                f"volatility {evaluation['volatility']}, seasonality: {evaluation['has_seasonality']}. "
      
       thought = self.llm(immediate, max_length=100, num_return_sequences=1)[0]['generated_text']
       print(f" Agent Thinking: {thought[:150]}...")
      
       if evaluation['has_seasonality']:
           self.selected_model = 'naive_seasonal'
           purpose = "Seasonality detected - utilizing Naive Seasonal mannequin"
       elif float(evaluation['volatility']) > 0.3:
           self.selected_model = 'exponential_smoothing'
           purpose = "High volatility - utilizing Exponential Smoothing"
       else:
           self.selected_model = 'linear_regression'
           purpose = "Stable development - utilizing Linear Regression"
      
       print(f" Decision: {purpose}")
       return self.selected_model
  
   def act(self, horizon=12):
       """Agent takes motion: trains mannequin and generates forecast"""
       print("n ACTION PHASE")
      
       practice, val = self.ts[:-12], self.ts[-12:]
      
       mannequin = self.fashions[self.selected_model]
       print(f" Training {self.selected_model}...")
       mannequin.match(practice)
      
       self.forecast = mannequin.predict(horizon)
      
       if len(val) > 0:
           val_pred = mannequin.predict(len(val))
           accuracy = 100 - mape(val, val_pred)
           print(f" Validation Accuracy: {accuracy:.2f}%")
      
       print(f" Generated {horizon}-step forecast")
       return self.forecast
  
   def clarify(self):
       """Agent explains its predictions"""
       print("n EXPLANATION PHASE")
      
       forecast_values = self.forecast.values().flatten()
       hist_values = self.ts.values().flatten()
      
       change = ((forecast_values[-1] - hist_values[-1]) / hist_values[-1]) * 100
       route = "enhance" if change > 0 else "lower"
      
       clarification = f"Based on my evaluation utilizing {self.selected_model}, " 
                    f"I predict a {abs(change):.1f}% {route} within the subsequent interval. " 
                    f"Forecast vary: {forecast_values.min():.2f} to {forecast_values.max():.2f}. " 
                    f"Historical imply was {hist_values.imply():.2f}."
      
       print(f" {clarification}")
      
       immediate = f"Forecast abstract: {clarification} Explain implications:"
       abstract = self.llm(immediate, max_length=120)[0]['generated_text']
       print(f"n Agent Summary: {abstract[:200]}...")
      
       return clarification
  
   def visualize(self):
       """Agent creates visualization of its work"""
       print("n Generating visualization...")
      
       plt.determine(figsize=(14, 6))
      
       self.ts.plot(label='Historical Data', lw=2)
      
       self.forecast.plot(label=f'Forecast ({self.selected_model})',
                         lw=2, linestyle='--')
      
       plt.title(' Agentic AI Time Series Forecast', fontsize=16, fontweight='daring')
       plt.xlabel('Date', fontsize=12)
       plt.ylabel('Value', fontsize=12)
       plt.legend(loc='finest', fontsize=11)
       plt.grid(True, alpha=0.3)
       plt.tight_layout()
       plt.present()

def create_sample_data():
   """Generate pattern time collection knowledge"""
   dates = pd.date_range(begin='2020-01-01', durations=48, freq='M')
   development = np.linspace(100, 150, 48)
   seasonality = 10 * np.sin(np.linspace(0, 4*np.pi, 48))
   noise = np.random.regular(0, 3, 48)
   values = development + seasonality + noise
  
   return pd.DataBody({'date': dates, 'worth': values})

def essential():
   """Main execution: Agent autonomously handles forecasting process"""
   print("="*70)
   print(" AGENTIC AI TIME SERIES FORECASTING SYSTEM")
   print("="*70)
  
   print("n Loading knowledge...")
   knowledge = create_sample_data()
   print(f"Loaded {len(knowledge)} knowledge factors from 2020-01 to 2023-12")
  
   agent = TimeSeriesAgent()
  
   evaluation = agent.understand(knowledge)
   agent.purpose(evaluation)
   agent.act(horizon=12)
   agent.clarify()
   agent.visualize()
  
   print("n" + "="*70)
   print(" AGENT COMPLETED FORECASTING TASK SUCCESSFULLY")
   print("="*70)


if __name__ == "__main__":
   essential()