Algorithm for maximizing shipping profit with limitations on mass and cost

You can try the framework optuna for hyperparameter tuning.

Here is an example code that you can try. Products are named product1 etc found in parameters.json file. Data values are just assumptions.

Study/optimization session are now saved in sqlite db. This will support interrupt and resume. See version log in the code.

parameters.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()

Output

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

If you increase the number of trials the program might be able to find a more profitable parameter values.

Another option is by using scipy. The sample below contains 3 products, that can be scaled of course. The constrains are the purchase and max truck mass capacity.

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

output

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