Algorithme de maximisation de l'expédition de profit avec des limites sur la masse et le coût

Vous pouvez essayer de le cadre optuna pour hyper-paramètre de réglage.

Voici un exemple de code que vous pouvez essayer. Les produits sont nommés produit1 etc trouvé dans les paramètres.fichier json. Les valeurs de données sont juste des hypothèses.

Étude/optimisation de session sont désormais enregistrés dans sqlite db. Cela permettra de soutenir l'interrompre et de reprendre. Voir la version du journal dans le code.

les paramètres.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
        }
    }
}

Code

"""
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()

Sortie

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

Si l'on augmente le nombre d'essais, le programme pourrait être en mesure de trouver un moyen plus rentable de valeurs de paramètre.

Une autre option est d'utiliser scipy. L'exemple ci-dessous contient les 3 produits, qui peuvent être mis à l'échelle, bien sûr. Les contraintes sont l'achat et max de camion de masse de la capacité.

code

"""
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()

sortie

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