Algoritmo per ottimizzare la spedizione di profitto con le limitazioni di massa e il costo

Si può provare il quadro optuna per hyperparameter tuning.

Ecco un esempio di codice che si può provare. I prodotti sono denominati prodotto1 etc trovato nei parametri.file json. I valori dei dati sono solo ipotesi.

Studio e ottimizzazione di sessione vengono salvati in un db sqlite. Questo permetterà di interrompere e riprendere. Vedi versione di registro nel codice.

i parametri.json

{
    "study_name": "st5_tpe",
    "sampler": "tpe",
    "trials": 1000,
    "max_purchase": 7000,
    "min_weight_no_cost": 1000,
    "high_weight_additional_cost": 0.5,
    "trucks": {
        "smalltruck": {
            "maxmass": 1000,
            "cost": 75
        },
        "mediumtruck": {
            "maxmass": 2000,
            "cost": 150
        },
        "bigtruck": {
            "maxmass": 5000,
            "cost": 400
        }
    },
    "products": {
        "product1_qty": {
            "min": 20,
            "max": 100,
            "massperunit": 2,
            "buyprice": 5,
            "sellprice": 8
        },
        "product2_qty": {
            "min": 20,
            "max": 100,
            "massperunit": 4,
            "buyprice": 6,
            "sellprice": 10
        },
        "product3_qty": {
            "min": 20,
            "max": 100,
            "massperunit": 1,
            "buyprice": 4,
            "sellprice": 6
        },
        "product4_qty": {
            "min": 20,
            "max": 100,
            "massperunit": 2,
            "buyprice": 7,
            "sellprice": 10
        },
        "product5_qty": {
            "min": 20,
            "max": 100,
            "massperunit": 2,
            "buyprice": 5,
            "sellprice": 8
        },
        "product6_qty": {
            "min": 20,
            "max": 100,
            "massperunit": 1,
            "buyprice": 5,
            "sellprice": 7
        },
        "product7_qty": {
            "min": 20,
            "max": 100,
            "massperunit": 1,
            "buyprice": 8,
            "sellprice": 12
        }
    }
}

Codice

"""
shipping_trading.py


version 0.7.0
    * Calculate and show ROI (return of investment) and other info.
    * Add user attribute to get other costs.
    * Raise exception when max_purchase key is missing in parameters.json file.
    * Continue the study even when trucks key is missing in parameters.json file.
    
version 0.6.0
    * Save study/optimization session in sqlite db, with this it can now supports interrupt and resume.
      When study session is interrupted it can be resumed later using data from previous session.
    * Add study_name key in parameters.json file. Sqlite db name is based on study_name. If you
      want new study/optimization session, modify the study_name. If you are re-running the
      same study_name, it will run and continue from previous session. Example:
      study_name=st8, sqlite_dbname=mydb_st8.db
      By default study_name is example_study when you remove study_name key in parameters.json file.
    * Remove printing in console on truck info.

version 0.5.0
    * Replace kg with qty in parameters.json file.
    * Add massperunit in the product.
    * Optimize qty not mass.
    * Refactor

version 0.4.0
    * Add truck size optimization. It is contrained by the cost of using truck as well as the max kg capacity.
      The optimizer may suggest a medium instead of a big truck if profit is higher as big truck is expensive.
      profit = profit - truck_cost - other_costs
    * Modify parameters.json file, trucks key is added.

version 0.3.0
    * Read sampler, and number of trials from parameters.json file.
      User inputs can now be processed from that file.

version 0.2.0
    * Read a new parameters.json format.
    * Refactor get_parameters().

version 0.1.0
    * Add additional cost if total product weight is high.
"""


__version__ = '0.7.0'


import json

import optuna


def get_parameters():
    """
    Read parameters.json file to get the parameters to optimize, etc.
    """
    fn = 'parameters.json'
    products, trucks = {}, {}

    with open(fn) as json_file:
        values = json.load(json_file)

        max_purchase = values.get('max_purchase', None)
        if max_purchase is None:
            raise Exception('Missing max_purchase, please specify max_purchase in json file, i.e "max_purchase": 1000')

        study_name = values.get('study_name', "example_study")
        sampler = values.get('sampler', "tpe")
        trials = values.get('trials', 100)
        min_weight_no_cost = values.get('min_weight_no_cost', None)
        high_weight_additional_cost = values.get('high_weight_additional_cost', None)
        products = values.get('products', None)
        trucks = values.get('trucks', None)

    return (products, trucks, sampler, trials, max_purchase, min_weight_no_cost, high_weight_additional_cost, study_name)


def objective(trial):
    """
    Maximize profit.
    """
    gp = get_parameters()
    (products, trucks, _, _, max_purchase,
        min_weight_no_cost, high_weight_additional_cost, _) = gp

    # Ask the optimizer the product qty to use try.
    new_param = {}    
    for k, v in products.items():
        suggested_value = trial.suggest_int(k, v['min'], v['max'])  # get suggested value from sampler
        new_param.update({k: {'suggested': suggested_value,
                               'massperunit': v['massperunit'],
                               'buyprice': v['buyprice'],
                               'sellprice': v['sellprice']}})

    # Ask the sampler which truck to use, small, medium ....
    truck_max_wt, truck_cost = None, None
    if trucks is not None:
        truck = trial.suggest_categorical("truck", list(trucks.keys()))

        # Define truck limits based on suggested truck size.
        truck_max_wt = trucks[truck]['maxmass']
        truck_cost = trucks[truck]['cost']

    # If total wt or total amount is exceeded, we return a 0 profit.
    total_wt, total_buy, profit = 0, 0, 0
    for k, v in new_param.items():
        total_wt += v['suggested'] * v['massperunit']
        total_buy += v['suggested'] * v['buyprice']
        profit += v['suggested'] * (v['sellprice'] - v['buyprice'])

    # (1) Truck mass limit
    if truck_max_wt is not None:
        if total_wt > truck_max_wt:
            return 0

    # (2) Purchase limit amount
    if max_purchase is not None:
        if total_buy > max_purchase:
            return 0

    # Cost for higher transport weight
    cost_high_weight = 0
    if min_weight_no_cost is not None and high_weight_additional_cost is not None:
        excess_weight = total_wt - min_weight_no_cost
        if excess_weight > 0:
            cost_high_weight += (total_wt - min_weight_no_cost) * high_weight_additional_cost

    # Cost for using a truck, can be small, medium etc.
    cost_truck_usage = 0
    if truck_cost is not None:
        cost_truck_usage += truck_cost

    # Total cost
    other_costs = cost_high_weight + cost_truck_usage
    trial.set_user_attr("other_costs", other_costs)

    # Adjust profit
    profit = profit - other_costs

    # Send this profit to optimizer so that it will consider this value
    # in its optimization algo and would suggest a better value next time we ask again.
    return profit


def return_of_investment(study, products):
    """
    Returns ROI.

    ROI = Return Of Investment
    ROI = 100 * profit/costs
    """
    product_sales, product_costs = 0, 0
    for (k, v), (k1, v1) in zip(products.items(), study.best_params.items()):
        if k == 'truck':
            continue
        assert k == k1
        product_sales += v1 * v['sellprice']
        product_costs += v1 * v['buyprice']
        
    other_costs = study.best_trial.user_attrs['other_costs']
    total_costs = product_costs + other_costs

    calculated_profit = product_sales - total_costs
    study_profit = study.best_trial.values[0]
    assert calculated_profit == study_profit
    
    return_of_investment = 100 * calculated_profit/total_costs

    return return_of_investment, product_sales, product_costs, other_costs


def main():
    # Read parameters.json file for user data input.
    gp = get_parameters()
    (products, trucks, optsampler, num_trials,
        max_purchase, _, _, study_name) = gp

    # Location of sqlite db where optimization session data are saved.
    sqlite_dbname = f'sqlite:///mydb_{study_name}.db'

    # Available samplers to use:
    # https://optuna.readthedocs.io/en/stable/reference/samplers.html
    # https://optuna.readthedocs.io/en/stable/reference/generated/optuna.integration.SkoptSampler.html
    # https://optuna.readthedocs.io/en/stable/reference/generated/optuna.integration.BoTorchSampler.html
    if optsampler.lower() == 'cmaes':
        sampler = optuna.samplers.CmaEsSampler(n_startup_trials=1, seed=100)
    elif optsampler.lower() == 'tpe':
        sampler = optuna.samplers.TPESampler(n_startup_trials=10, multivariate=False, group=False, seed=100, n_ei_candidates=24)
    else:
        print(f'Warning, {optsampler} is not supported, we will be using tpe sampler instead.')
        optsampler = 'tpe'
        sampler = optuna.samplers.TPESampler(n_startup_trials=10, multivariate=False, group=False, seed=100, n_ei_candidates=24)

    # Store optimization in storage and supports interrupt/resume.
    study = optuna.create_study(storage=sqlite_dbname, sampler=sampler, study_name=study_name, load_if_exists=True, direction='maximize')
    study.optimize(objective, n_trials=num_trials)

    # Show summary and best parameter values to maximize profit.
    print()
    print(f'study_name: {study_name}')
    print(f'sqlite dbname: {sqlite_dbname}')
    print(f'sampler: {optsampler}')
    print(f'trials: {num_trials}')
    print()

    print(f'Max Purchase Amount: {max_purchase}')
    print()

    print('Products being optimized:')
    for k, v in products.items():
        print(f'{k}: {v}')
    print()

    if trucks is not None:
        print('Trucks being optimized:')
        for k, v in trucks.items():
            print(f'{k}: {v}')
        print()

    print('Study/Optimization results:')
    objective_name = 'profit'
    print(f'best parameter value : {study.best_params}')
    print(f'best value           : {study.best_trial.values[0]}')
    print(f'best trial           : {study.best_trial.number}')
    print(f'objective            : {objective_name}')
    print()

    # Show other info like roi, etc.
    roi, product_sales, product_costs, other_costs = return_of_investment(study, products)
    print('Other info.:')    
    print(f'Return Of Investment : {roi:0.2f}%, profit/costs')
    print(f'Product Sales        : {product_sales:0.2f}')
    print(f'Product Costs        : {product_costs:0.2f}')
    print(f'Other Costs          : {other_costs:0.2f}')
    print(f'Total Costs          : {product_costs + other_costs:0.2f}')
    print(f'Profit               : {product_sales - (product_costs + other_costs):0.2f}')
    print(f'Capital              : {max_purchase:0.2f}')
    print(f'Total Spent          : {product_costs + other_costs:0.2f} ({100*(product_costs + other_costs)/max_purchase:0.2f}% of Capital)')
    print(f'Capital Balance      : {max_purchase - product_costs - other_costs:0.2f}')
    print()


if __name__ == '__main__':
    main()

Uscita

study_name: st5_tpe
sqlite dbname: sqlite:///mydb_st5_tpe.db
sampler: tpe
trials: 1000

Max Purchase Amount: 7000

Products being optimized:
product1_qty: {'min': 20, 'max': 100, 'massperunit': 2, 'buyprice': 5, 'sellprice': 8}
product2_qty: {'min': 20, 'max': 100, 'massperunit': 4, 'buyprice': 6, 'sellprice': 10}
product3_qty: {'min': 20, 'max': 100, 'massperunit': 1, 'buyprice': 4, 'sellprice': 6}
product4_qty: {'min': 20, 'max': 100, 'massperunit': 2, 'buyprice': 7, 'sellprice': 10}
product5_qty: {'min': 20, 'max': 100, 'massperunit': 2, 'buyprice': 5, 'sellprice': 8}
product6_qty: {'min': 20, 'max': 100, 'massperunit': 1, 'buyprice': 5, 'sellprice': 7}
product7_qty: {'min': 20, 'max': 100, 'massperunit': 1, 'buyprice': 8, 'sellprice': 12}

Trucks being optimized:
smalltruck: {'maxmass': 1000, 'cost': 75}
mediumtruck: {'maxmass': 2000, 'cost': 150}
bigtruck: {'maxmass': 5000, 'cost': 400}

Study/Optimization results:
best parameter value : {'product1_qty': 99, 'product2_qty': 96, 'product3_qty': 93, 'product4_qty': 96, 'product5_qty': 100, 'product6_qty': 100, 'product7_qty': 100, 'truck': 'mediumtruck'}
best value           : 1771.5
best trial           : 865
objective            : profit

Other info.:
Return Of Investment : 42.19%, profit/costs
Product Sales        : 5970.00
Product Costs        : 3915.00
Other Costs          : 283.50
Total Costs          : 4198.50
Profit               : 1771.50
Capital              : 7000.00
Total Spent          : 4198.50 (59.98% of Capital)
Capital Balance      : 2801.50

Se si aumenta il numero di prove, il programma potrebbe essere in grado di trovare un più redditizio valori di parametro.

Un'altra opzione è usare scipy. Nell'esempio riportato di seguito contiene 3 prodotti, che possono essere scalati, naturalmente. I vincoli sono l'acquisto e max camion di massa capacità.

codice

"""
shipping_trading_solver.py

Ref: https://docs.scipy.org/doc/scipy/reference/generated/scipy.optimize.minimize.html#scipy.optimize.minimize
"""


from scipy.optimize import minimize


# Constants
sellprice = [8, 7, 10]
buyprice = [6, 5, 6]
mass_per_unit = [1, 2, 3]

purchase_limit = 100
truck_mass_limit = 70


def objective(x):
    """
    objective, return value as negative to maximize.
    x: quantity
    """
    profit = 0
    for (v, s, b) in zip(x, sellprice, buyprice):
        profit += v * (s - b)

    return -profit


def purchase_cons(x):
    """
    Used for constrain
    x: quantity
    """
    purchases = 0
    for (v, b) in zip(x, buyprice):
        purchases += v * b
    
    return purchase_limit - purchases  # not negative


def mass_cons(x):
    """
    Used for constrain
    mass = qty * mass/qty
    x: quantity
    """
    mass = 0
    for (v, m) in zip(x, mass_per_unit):
        mass += v * m
    
    return truck_mass_limit - mass  # not negative


def profit_cons(x):
    """
    Used for constrain
    x: quantity
    """
    profit = 0
    for (v, s, b) in zip(x, sellprice, buyprice):
        profit += v * (s - b)

    return profit  # not negative


def main():
    # Define constrained. Note: ineq=non-negative, eq=zero
    cons = (
        {'type': 'ineq', 'fun': purchase_cons},
        {'type': 'ineq', 'fun': mass_cons},
        {'type': 'ineq', 'fun': profit_cons}
    )

    # Bounds of product quantity, (min,max)
    bound = ((0, 50), (0, 20), (0, 30))

    # Initial values
    init_values = (0, 0, 0)

    # Start minimizing
    # SLSQP = Sequential Least Squares Programming
    res = minimize(objective, init_values, method='SLSQP', bounds=bound, constraints=cons)

    # Show summary
    print('Results summary:')
    print(f'optimization message: {res.message}')
    print(f'sucess status: {res.success}')
    print(f'profit: {-res.fun:0.2f}')
    print(f'best param values: {[round(v, 5) for v in res.x]}')
    print()

    # Verify results
    print('Verify purchase and mass limits:')

    # (1) purchases
    total_purchases = 0
    for (qty, b) in zip(res.x, buyprice):
        total_purchases += qty * b
    print(f'actual total_purchases: {total_purchases:0.0f}, purchase_limit: {purchase_limit}')

    # (2) mass
    total_mass = 0    
    for (qty, m) in zip(res.x, mass_per_unit):
        total_mass += qty * m
    print(f'actual total_mass: {total_mass:0.0f}, truck_mass_limit: {truck_mass_limit}')


if __name__ == '__main__':
    main()

uscita

Results summary:
optimization message: Optimization terminated successfully
sucess status: True
profit: 66.67
best param values: [0.0, 0.0, 16.66667]

Verify purchase and mass limits:
actual total_purchases: 100, purchase_limit: 100
actual total_mass: 50, truck_mass_limit: 70