Struct Service 

pub struct Service { /* private fields */ }

Genetic algorithm optimization service that manages the entire optimization lifecycle. Handles request creation, population generation, genotype evaluation, and breeding.

Implementations

impl Service

pub async fn new_optimization_request( &self, type_name: &str, type_hash: i32, goal: FitnessGoal, schedule: Schedule, selector: Selector, mutagen: Mutagen, crossover: Crossover, distribution: Distribution, data: Option<impl Serialize + Send + Sync>, ) -> Result<Uuid, Error>

Creates a new genetic algorithm optimization request.

This is the main entry point for starting genetic algorithm optimization. It configures and initiates an evolutionary process that will attempt to find optimal solutions for your problem type.

Parameters

• type_name - Human-readable name of the type being optimized (from Encodeable::NAME)
• type_hash - Unique hash identifying the type structure (from Encodeable::HASH)
• goal - Optimization objective and stopping criteria
• schedule - Controls generation timing and population lifecycle
• selector - Parent selection strategy for breeding operations
• mutagen - Mutation behavior including temperature and rate schedules
• crossover - Genetic recombination method for combining parents
• distribution - Initial population generation strategy

Optimization Goal

The goal parameter defines when optimization should stop:

• FitnessGoal::maximize(0.95) - Stop when fitness reaches 95% or higher
• FitnessGoal::minimize(0.1) - Stop when fitness drops to 10% or lower

Population Management

The schedule controls generation timing and resource allocation:

• Schedule::generational(100, 50) - 100 generations, 50 individuals each
• Schedule::rolling(5000, 100, 10) - 5000 total evaluations, 100 max population, breed 10 at a time

Parent Selection

The selector determines how parents are chosen for breeding:

• Selector::tournament(3, 100) - Tournament selection (size 3) from 100 candidates
• Selector::roulette(100) - Fitness-proportionate selection from 100 candidates

Mutation Strategy

The mutagen combines temperature (step size) and mutation rate (frequency):

• Mutagen::constant(0.5, 0.3) - Fixed 50% temperature, 30% mutation rate
• Adaptive strategies use Temperature::linear() and MutationRate::exponential()

Genetic Recombination

The crossover method combines genetic material from parents:

• Crossover::uniform(0.5) - Each gene has 50% chance from first parent
• Crossover::single_point() - Cut genome at random point, swap tails

Initial Population

The distribution strategy affects initial diversity and convergence:

• Distribution::latin_hypercube(50) - Structured sampling for better coverage
• Distribution::random(50) - Pure random sampling

Common Configurations

Quick Convergence (Time-Constrained)

use fx_durable_ga::models::*;

service.new_optimization_request(
    MyType::NAME,
    MyType::HASH,
    FitnessGoal::maximize(0.90)?,                    // Modest target
    Schedule::generational(20, 30),                 // Small, fast generations
    Selector::tournament(5, 50)?,                    // Strong selection pressure
    Mutagen::new(
        Temperature::exponential(0.8, 0.1, 1.2, 3)?, // Rapid cooling
        MutationRate::exponential(0.6, 0.05, 1.1, 2)?
    ),
    Crossover::uniform(0.6)?,                       // High recombination
    Distribution::latin_hypercube(30),
    None::<()>,
).await?;

Balanced Search (General Purpose)

use fx_durable_ga::models::*;

service.new_optimization_request(
    MyType::NAME,
    MyType::HASH,
    FitnessGoal::maximize(0.95)?,                    // High-quality target
    Schedule::generational(50, 50),                 // Moderate generations
    Selector::tournament(3, 100)?,                   // Balanced selection
    Mutagen::new(
        Temperature::linear(0.7, 0.2, 1.0)?,         // Gradual cooling
        MutationRate::linear(0.4, 0.1, 1.0)?
    ),
    Crossover::uniform(0.5)?,                       // Balanced recombination
    Distribution::latin_hypercube(50),
    None::<()>,
).await?;

Thorough Exploration (Complex Problems)

use fx_durable_ga::models::*;

service.new_optimization_request(
    MyType::NAME,
    MyType::HASH,
    FitnessGoal::maximize(0.99)?,                    // High precision target
    Schedule::rolling(10000, 200, 20),              // Large budget
    Selector::tournament(2, 150)?,                   // Low selection pressure
    Mutagen::constant(0.6, 0.4)?,                   // Sustained exploration
    Crossover::single_point(),                      // Conservative recombination
    Distribution::latin_hypercube(100),             // High initial diversity
    None::<()>,
).await?;

Parameter Interactions

Key relationships to consider:

• High selection pressure + Low mutation: Fast convergence, may get stuck
• Low selection pressure + High mutation: Slow convergence, good exploration
• Large populations + Tournament selection: Better genetic diversity
• Small populations + Roulette selection: Risk of premature convergence
• Exponential decay + Generational schedule: Classic simulated annealing
• Linear decay + Rolling schedule: Steady refinement over time

Returns

Returns Ok(()) if the optimization request was successfully created and scheduled. The actual optimization runs asynchronously via the event/job system.

Errors

• Returns error if the specified type hasn’t been registered with an evaluator
• Returns error if database operations fail
• Parameter validation errors are caught at construction time (e.g., FitnessGoal::maximize(1.5) fails)

Example Usage

use fx_durable_ga::models::*;

// Define your problem type
#[derive(Debug)]
struct Point { x: f64, y: f64 }

impl Encodeable for Point {
    const NAME: &'static str = "Point";
    type Phenotype = Point;
    fn morphology() -> Vec<GeneBounds> {
        vec![
            GeneBounds::decimal(0.0, 10.0, 1000, 3).unwrap(),
            GeneBounds::decimal(0.0, 10.0, 1000, 3).unwrap(),
        ]
    }
    fn encode(&self) -> Vec<i64> {
        let bounds = Self::morphology();
        vec![
            bounds[0].from_sample(self.x / 10.0),
            bounds[1].from_sample(self.y / 10.0),
        ]
    }
    fn decode(genes: &[i64]) -> Self::Phenotype {
        let bounds = Self::morphology();
        Point {
            x: bounds[0].decode_f64(genes[0]),
            y: bounds[1].decode_f64(genes[1]),
        }
    }
}

// Start optimization
service.new_optimization_request(
    Point::NAME,
    Point::HASH,
    FitnessGoal::maximize(0.95)?,
    Schedule::generational(50, 50),
    Selector::tournament(3, 100)?,
    Mutagen::constant(0.5, 0.3)?,
    Crossover::uniform(0.5)?,
    Distribution::latin_hypercube(50),
    None::<()>,
).await?;

Examples found in repository

examples/regression_model.rs (lines 241-251)

async fn main() -> Result<()> {
    dotenv::from_filename(".env.local").ok();
    tracing_subscriber::fmt()
        .pretty()
        .with_thread_ids(true)
        .with_max_level(Level::INFO)
        .init();

    let database_url = env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(&database_url)
        .await?;

    // Run all default migrations
    fx_durable_ga::migrations::run_default_migrations(&pool).await?;

    let service = Arc::new(
        bootstrap(pool.clone())
            .await?
            .register::<NeuralArchitecture, _>(ArchitectureEvaluator)
            .await?
            .build(),
    );

    let mut registry = fx_event_bus::EventHandlerRegistry::new();
    register_event_handlers(
        Arc::new(Queries::new(FX_MQ_JOBS_SCHEMA_NAME)),
        service.clone(),
        &mut registry,
    );
    let mut listener = fx_event_bus::Listener::new(pool.clone(), registry);
    tokio::spawn(async move { listener.listen(None).await });

    let host_id = Uuid::parse_str("00000000-0000-0000-0000-123456789abc")?;
    let mut jobs_listener = fx_mq_jobs::Listener::new(
        pool.clone(),
        register_job_handlers(&service, fx_mq_jobs::RegistryBuilder::new()),
        WORKERS,
        host_id,
        Duration::from_secs(600),
    )
    .await?;
    tokio::spawn(async move { jobs_listener.listen().await });

    service
        .new_optimization_request(
            NeuralArchitecture::NAME,
            NeuralArchitecture::HASH,
            FitnessGoal::minimize(FITNESS_TARGET)?,
            Schedule::generational(10, 10),
            Selector::tournament(5, 15)?,
            Mutagen::new(Temperature::constant(0.8)?, MutationRate::constant(0.4)?),
            Crossover::uniform(0.5)?,
            Distribution::latin_hypercube(15),
            None::<()>,
        )
        .await?;

    tokio::time::sleep(Duration::from_secs(3600)).await;
    Ok(())
}

examples/point_search.rs (lines 195-205)

async fn main() -> Result<()> {
    dotenv::from_filename(".env.local").ok();

    // Initialize logging to see optimization progress
    tracing_subscriber::fmt()
        .with_max_level(tracing::Level::INFO)
        .init();

    // Database setup - genetic algorithms need persistent storage for populations
    let database_url = env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let pool = PgPoolOptions::new()
        .max_connections(50)
        .connect(&database_url)
        .await?;

    fx_event_bus::run_migrations(&pool).await?;
    fx_mq_jobs::run_migrations(&pool, fx_mq_jobs::FX_MQ_JOBS_SCHEMA_NAME).await?;

    // Bootstrap the optimization service and register our problem type
    let target_point = Point {
        x: 1.0,
        y: 1.5,
        z: 2.5,
    }; // Target point to find
    let evaluator = PointDistanceEvaluator { target_point };
    let service = Arc::new(
        bootstrap(pool.clone())
            .await?
            .register::<Point, _>(evaluator)
            .await?
            .build(),
    );

    // setup event handling and spawn an event handling agent
    let mut registry = fx_event_bus::EventHandlerRegistry::new();
    register_event_handlers(
        Arc::new(Queries::new(FX_MQ_JOBS_SCHEMA_NAME)),
        service.clone(),
        &mut registry,
    );
    let mut listener = fx_event_bus::Listener::new(pool.clone(), registry);
    let _events_handle = tokio::spawn(async move {
        listener.listen(None).await?;
        Ok::<(), sqlx::Error>(())
    });

    // setup job handling and initiate workers
    let host_id = Uuid::parse_str("ba3a4752-c4ce-4129-aa1d-55a0b2107e68").expect("valid uuid");
    let hold_for = Duration::from_secs(TIMEOUT_SECONDS);
    let mut jobs_listener = fx_mq_jobs::Listener::new(
        pool.clone(),
        register_job_handlers(&service, fx_mq_jobs::RegistryBuilder::new()),
        8,
        host_id,
        hold_for,
    )
    .await?;
    let _jobs_handle = tokio::spawn(async move {
        jobs_listener.listen().await?;
        Ok::<(), anyhow::Error>(())
    });

    // Create multiple optimization requests with different genetic algorithm parameters
    for _ in 0..1 {
        service
            .new_optimization_request(
                Point::NAME,
                Point::HASH,
                FitnessGoal::minimize(0.01)?, // Stop when distance ≤ 0.01
                Schedule::generational(200, 30),
                Selector::tournament(7, 100)?,
                Mutagen::new(Temperature::constant(0.7)?, MutationRate::constant(0.3)?),
                Crossover::uniform(0.5)?,
                Distribution::latin_hypercube(1000),
                None::<()>,
            )
            .await?;
    }

    let timeout_duration = Duration::from_secs(TIMEOUT_SECONDS);
    let start_time = std::time::Instant::now();

    loop {
        tokio::time::sleep(Duration::from_secs(1)).await;

        // Check if we've exceeded the timeout
        if start_time.elapsed() >= timeout_duration {
            println!(
                "Timeout reached after {} seconds. Stopping optimization.",
                timeout_duration.as_secs()
            );
            break;
        }
    }

    Ok(())
}

examples/feature_engineering.rs (lines 423-433)

async fn main() -> Result<()> {
    dotenv::from_filename(".env.local").ok();
    tracing_subscriber::fmt()
        .pretty()
        .with_thread_ids(true)
        .with_max_level(Level::INFO)
        .init();

    let database_url = env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(&database_url)
        .await?;

    // Run all default migrations
    fx_durable_ga::migrations::run_default_migrations(&pool).await?;

    let service = Arc::new(
        bootstrap(pool.clone())
            .await?
            .register::<FeatureConfig, _>(FeatureEvaluator)
            .await?
            .build(),
    );

    let mut registry = fx_event_bus::EventHandlerRegistry::new();
    register_event_handlers(
        Arc::new(Queries::new(FX_MQ_JOBS_SCHEMA_NAME)),
        service.clone(),
        &mut registry,
    );
    let mut listener = fx_event_bus::Listener::new(pool.clone(), registry);
    tokio::spawn(async move { listener.listen(None).await });

    let host_id = Uuid::parse_str("00000000-0000-0000-0000-123456789abc")?;
    let mut jobs_listener = fx_mq_jobs::Listener::new(
        pool.clone(),
        register_job_handlers(&service, fx_mq_jobs::RegistryBuilder::new()),
        WORKERS,
        host_id,
        Duration::from_secs(600),
    )
    .await?;
    tokio::spawn(async move { jobs_listener.listen().await });

    let request_id = service
        .new_optimization_request(
            FeatureConfig::NAME,
            FeatureConfig::HASH,
            FitnessGoal::minimize(FITNESS_TARGET)?,
            Schedule::generational(40, 10),
            Selector::tournament(5, 45)?,
            Mutagen::new(Temperature::constant(0.7)?, MutationRate::constant(0.35)?),
            Crossover::uniform(0.5)?,
            Distribution::latin_hypercube(40),
            None::<()>,
        )
        .await?;

    // Poll for completion every 15 seconds (timeout after 1 hour)
    let timeout = Duration::from_secs(7200);
    let poll_interval = Duration::from_secs(15);
    let start = std::time::Instant::now();

    loop {
        if service.is_request_concluded(request_id).await? {
            println!("\nOptimization completed!");
            break;
        }

        if start.elapsed() > timeout {
            println!("\nOptimization timed out after 1 hour.");
            break;
        }

        tokio::time::sleep(poll_interval).await;
    }

    // Get and print the best configuration
    if let Some((genotype, fitness)) = service.get_best_genotype(request_id).await? {
        let config = FeatureConfig::decode(&genotype.genome());
        println!("\n=== Best Configuration ===");
        println!("Fitness (MSE): {:.6}", fitness);
        println!("RMSE: {:.6}°C", fitness.sqrt());
        println!("\nHyperparameters:");
        println!("  Hidden Size: {}", config.hidden_size);
        println!("  Learning Rate: {}", config.learning_rate);
        println!("  Sequence Length: {}", config.sequence_length);
        println!("\nFeatures:");
        for (i, feature) in config.features.iter().enumerate() {
            let pipeline = feature
                .transforms
                .iter()
                .map(|t| t.to_string())
                .collect::<Vec<_>>()
                .join(" → ");
            if pipeline.is_empty() {
                println!("  feat_{}: {}", i, feature.source);
            } else {
                println!("  feat_{}: {} → {}", i, feature.source, pipeline);
            }
        }
    } else {
        println!("No genotypes were evaluated.");
    }

    Ok(())
}

pub async fn interrupt_request(&self, request_id: Uuid) -> Result<(), Error>

Interrupts a running optimization request. Publishes an interruption event that concludes the request with the “Interrupted” reason.

pub async fn get_best_genotype( &self, request_id: Uuid, ) -> Result<Option<(Genotype, f64)>, Error>

Gets the best genotype for a request based on fitness. Returns the genotype with the best (min or max) fitness depending on the goal.

Examples found in repository

examples/feature_engineering.rs (line 456)

async fn main() -> Result<()> {
    dotenv::from_filename(".env.local").ok();
    tracing_subscriber::fmt()
        .pretty()
        .with_thread_ids(true)
        .with_max_level(Level::INFO)
        .init();

    let database_url = env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(&database_url)
        .await?;

    // Run all default migrations
    fx_durable_ga::migrations::run_default_migrations(&pool).await?;

    let service = Arc::new(
        bootstrap(pool.clone())
            .await?
            .register::<FeatureConfig, _>(FeatureEvaluator)
            .await?
            .build(),
    );

    let mut registry = fx_event_bus::EventHandlerRegistry::new();
    register_event_handlers(
        Arc::new(Queries::new(FX_MQ_JOBS_SCHEMA_NAME)),
        service.clone(),
        &mut registry,
    );
    let mut listener = fx_event_bus::Listener::new(pool.clone(), registry);
    tokio::spawn(async move { listener.listen(None).await });

    let host_id = Uuid::parse_str("00000000-0000-0000-0000-123456789abc")?;
    let mut jobs_listener = fx_mq_jobs::Listener::new(
        pool.clone(),
        register_job_handlers(&service, fx_mq_jobs::RegistryBuilder::new()),
        WORKERS,
        host_id,
        Duration::from_secs(600),
    )
    .await?;
    tokio::spawn(async move { jobs_listener.listen().await });

    let request_id = service
        .new_optimization_request(
            FeatureConfig::NAME,
            FeatureConfig::HASH,
            FitnessGoal::minimize(FITNESS_TARGET)?,
            Schedule::generational(40, 10),
            Selector::tournament(5, 45)?,
            Mutagen::new(Temperature::constant(0.7)?, MutationRate::constant(0.35)?),
            Crossover::uniform(0.5)?,
            Distribution::latin_hypercube(40),
            None::<()>,
        )
        .await?;

    // Poll for completion every 15 seconds (timeout after 1 hour)
    let timeout = Duration::from_secs(7200);
    let poll_interval = Duration::from_secs(15);
    let start = std::time::Instant::now();

    loop {
        if service.is_request_concluded(request_id).await? {
            println!("\nOptimization completed!");
            break;
        }

        if start.elapsed() > timeout {
            println!("\nOptimization timed out after 1 hour.");
            break;
        }

        tokio::time::sleep(poll_interval).await;
    }

    // Get and print the best configuration
    if let Some((genotype, fitness)) = service.get_best_genotype(request_id).await? {
        let config = FeatureConfig::decode(&genotype.genome());
        println!("\n=== Best Configuration ===");
        println!("Fitness (MSE): {:.6}", fitness);
        println!("RMSE: {:.6}°C", fitness.sqrt());
        println!("\nHyperparameters:");
        println!("  Hidden Size: {}", config.hidden_size);
        println!("  Learning Rate: {}", config.learning_rate);
        println!("  Sequence Length: {}", config.sequence_length);
        println!("\nFeatures:");
        for (i, feature) in config.features.iter().enumerate() {
            let pipeline = feature
                .transforms
                .iter()
                .map(|t| t.to_string())
                .collect::<Vec<_>>()
                .join(" → ");
            if pipeline.is_empty() {
                println!("  feat_{}: {}", i, feature.source);
            } else {
                println!("  feat_{}: {} → {}", i, feature.source, pipeline);
            }
        }
    } else {
        println!("No genotypes were evaluated.");
    }

    Ok(())
}

pub async fn is_request_concluded( &self, request_id: Uuid, ) -> Result<bool, Error>

Checks if an optimization request has concluded. Returns true if the request has been completed or terminated.

Examples found in repository

examples/feature_engineering.rs (line 442)

async fn main() -> Result<()> {
    dotenv::from_filename(".env.local").ok();
    tracing_subscriber::fmt()
        .pretty()
        .with_thread_ids(true)
        .with_max_level(Level::INFO)
        .init();

    let database_url = env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(&database_url)
        .await?;

    // Run all default migrations
    fx_durable_ga::migrations::run_default_migrations(&pool).await?;

    let service = Arc::new(
        bootstrap(pool.clone())
            .await?
            .register::<FeatureConfig, _>(FeatureEvaluator)
            .await?
            .build(),
    );

    let mut registry = fx_event_bus::EventHandlerRegistry::new();
    register_event_handlers(
        Arc::new(Queries::new(FX_MQ_JOBS_SCHEMA_NAME)),
        service.clone(),
        &mut registry,
    );
    let mut listener = fx_event_bus::Listener::new(pool.clone(), registry);
    tokio::spawn(async move { listener.listen(None).await });

    let host_id = Uuid::parse_str("00000000-0000-0000-0000-123456789abc")?;
    let mut jobs_listener = fx_mq_jobs::Listener::new(
        pool.clone(),
        register_job_handlers(&service, fx_mq_jobs::RegistryBuilder::new()),
        WORKERS,
        host_id,
        Duration::from_secs(600),
    )
    .await?;
    tokio::spawn(async move { jobs_listener.listen().await });

    let request_id = service
        .new_optimization_request(
            FeatureConfig::NAME,
            FeatureConfig::HASH,
            FitnessGoal::minimize(FITNESS_TARGET)?,
            Schedule::generational(40, 10),
            Selector::tournament(5, 45)?,
            Mutagen::new(Temperature::constant(0.7)?, MutationRate::constant(0.35)?),
            Crossover::uniform(0.5)?,
            Distribution::latin_hypercube(40),
            None::<()>,
        )
        .await?;

    // Poll for completion every 15 seconds (timeout after 1 hour)
    let timeout = Duration::from_secs(7200);
    let poll_interval = Duration::from_secs(15);
    let start = std::time::Instant::now();

    loop {
        if service.is_request_concluded(request_id).await? {
            println!("\nOptimization completed!");
            break;
        }

        if start.elapsed() > timeout {
            println!("\nOptimization timed out after 1 hour.");
            break;
        }

        tokio::time::sleep(poll_interval).await;
    }

    // Get and print the best configuration
    if let Some((genotype, fitness)) = service.get_best_genotype(request_id).await? {
        let config = FeatureConfig::decode(&genotype.genome());
        println!("\n=== Best Configuration ===");
        println!("Fitness (MSE): {:.6}", fitness);
        println!("RMSE: {:.6}°C", fitness.sqrt());
        println!("\nHyperparameters:");
        println!("  Hidden Size: {}", config.hidden_size);
        println!("  Learning Rate: {}", config.learning_rate);
        println!("  Sequence Length: {}", config.sequence_length);
        println!("\nFeatures:");
        for (i, feature) in config.features.iter().enumerate() {
            let pipeline = feature
                .transforms
                .iter()
                .map(|t| t.to_string())
                .collect::<Vec<_>>()
                .join(" → ");
            if pipeline.is_empty() {
                println!("  feat_{}: {}", i, feature.source);
            } else {
                println!("  feat_{}: {} → {}", i, feature.source, pipeline);
            }
        }
    } else {
        println!("No genotypes were evaluated.");
    }

    Ok(())
}

pub async fn search_genotypes( &self, filter: &Filter, limit: i64, ) -> Result<Vec<(Genotype, Option<f64>)>, Error>

Get a morphology by its type name

pub async fn get_phenotype<T: Encodeable>( &self, genotype_id: &Uuid, ) -> Result<T::Phenotype, Error>

Get a phenotype by its type name