2
\$\begingroup\$

Background: I'm a BI developer building a new dashboard for a client. They want to track performance for the week/month/year to date against the prior period. Unfortunately, I don't have direct access to the data source and I only have an extract containing two weeks of data. This means that most of my period over period metrics will just show zero, and that doesn't make for a good demo. Getting access to the data source or a complete extract isn't possible due to some organizational issues, so I decided to generate my own test data.

Data Set: This is a very simple data set covering performance at a set of warehouses. There are 4 fields in the output: Warehouse ID, Date, and two integer fields.

Approach: I wrote a python script to generate test data for a given range of dates, using statistics from the two-week data sample. I captured the average and standard deviation for daily rowcounts and the two measures needed for my analysis. My script takes these as inputs, and iterates through each date in the date range to generate mock data. The script outputs a separate file for each source warehouse, simulating the way data is extracted from the source system.

Questions: I'd like to get input on the following:

  1. I generate records one at a time as a list, then append the list to a dataframe. This seems inefficient but I couldn't find a better way to do it.(See line 99)
  2. Is there a better way to create test data from a sample?
  3. Are there any other ways for this code to be improved?

Code:

import pandas as pd
import random as rand
import math

# Set some parameters to create test data

Start_Date = '2021-01-01'
End_Date = '2023-06-28'

warehouses = [
    'A',
    'B',
    'C',
    'D',
    'E',
    'F',
    'G'
]

warehouse_json = {
    "A" : {
        "avg_rowcount" : 10,
        "stDev_rowcount": 2,
        "avg_CntQty": 17,
        "stDev_CntQty": 143,
        "avg_SysQty": 17,
        "stDev_SysQty": 143
    },
    "B" : {
        "avg_rowcount" : 10,
        "stDev_rowcount": 2,
        "avg_CntQty": 27,
        "stDev_CntQty": 193,
        "avg_SysQty": 27,
        "stDev_SysQty": 309
    },
    "C": {
        "avg_rowcount": 3,
        "stDev_rowcount": 2,
        "avg_CntQty": 50,
        "stDev_CntQty": 310,
        "avg_SysQty": 51,
        "stDev_SysQty": 3090
    },
    "D": {
        "avg_rowcount": 36,
        "stDev_rowcount": 99,
        "avg_CntQty": 22,
        "stDev_CntQty": 31,
        "avg_SysQty": 21,
        "stDev_SysQty": 31
    },
    "E": {
        "avg_rowcount": 35,
        "stDev_rowcount": 120,
        "avg_CntQty": 40,
        "stDev_CntQty": 116,
        "avg_SysQty": 40,
        "stDev_SysQty": 116
    },
    "F": {
        "avg_rowcount": 4,
        "stDev_rowcount": 2,
        "avg_CntQty": 89,
        "stDev_CntQty": 3352,
        "avg_SysQty": 88,
        "stDev_SysQty": 3359
    },
    "G": {
        "avg_rowcount": 2,
        "stDev_rowcount": 2,
        "avg_CntQty": 599,
        "stDev_CntQty": 28430,
        "avg_SysQty": 599,
        "stDev_SysQty": 28430
    }
}

#Make a date range to iterate through
date_range = pd.date_range(Start_Date, End_Date)

for warehouse_id in warehouses:
    # Create a blank data frame to hold test data
    df_test_data = pd.DataFrame(columns=["Warehouse_ID", "Date", "CntQty", "SysQty"])

    #Iterate through each date in the range created above
    for single_date in date_range:
            #Determine how many rows to create, using mean and stdev of observed rowcounts from the above json. Add 1 to ensure we generate at least one record.
            rows_to_generate = 1 + math.ceil(rand.normalvariate(warehouse_json[warehouse_id]["avg_rowcount"],warehouse_json[warehouse_id]["stDev_rowcount"]))
            #Generate the rows, plugging in the warehouse ID and date, and using rand to create values based on observed means and stdev
            for i in range(0, rows_to_generate):
                new_row = [
                    warehouse_id,
                    single_date,
                    abs(rand.normalvariate(warehouse_json[warehouse_id]["avg_CntQty"], warehouse_json[warehouse_id]["stDev_CntQty"])),
                    abs(rand.normalvariate(warehouse_json[warehouse_id]["avg_SysQty"], warehouse_json[warehouse_id]["stDev_SysQty"]))
                ]
                #Append the new row to the data frame created above. There's probably a better way to do this.
                df_test_data.loc[len(df_test_data)] = new_row

    #write the completed df to a csv
    output_path = '' + warehouse_id + 'TaskDetails.csv' #Filepath snipped out for privacy
    df_test_data.to_csv(output_path,index=False)

\$\endgroup\$
1
  • 1
    \$\begingroup\$ The daily quantities of an item in a warehouse are not independent random variables. The quantity today is likely to be correlated to the quantity yesterday. So it might make more sense to randomly generate the day-to-day changes rather than randomly generate the daily values directly. \$\endgroup\$
    – RootTwo
    Jun 30, 2023 at 7:05

1 Answer 1

3
\$\begingroup\$

Forgive my saying so, but

  • Clipping a normal distribution to have a minimum of 0 introduces significant statistical bias and doesn't make much sense;
  • ceil() also introduces bias; and
  • Applying an abs() makes far less sense and will produce a bizarre distribution indeed.

In the latter case, at the very least, demote that to a clip(); but with this approach this program will be a lie. Your target means will not be the means of tendency in its current form. There are options to get around this, each requiring a lie of some kind:

  • use the clipped normal and add a maximum twice the mean, which achieves the same mean but probably won't have the same standard deviation and won't be exactly normal;
  • use a uniform distribution instead with bounds chosen to get your same mean and standard deviation (which is only possible if the difference between your mean and minimum is less than sqrt 3 * your stdev); or
  • use a non-normal distribution such as lognorm, which achieves the same mean and standard deviation but will lose the normality you attempted.

Delete all of your code and replace it with a vectorised version that uses Numpy and Pandas. Don't iterate over each row.

When generating test data for most applications, it's very important that you set a constant (or at least explicitly-determined) random seed.

Clipped normals

import numpy as np
import pandas as pd

from numpy.random import default_rng


rand = default_rng(seed=0)
start_date = '2021-01-01'
end_date = '2023-06-28'

warehouse_data = {
    "A": {
        "avg_rowcount": 10,
        "stDev_rowcount": 2,
        "avg_CntQty": 17,
        "stDev_CntQty": 143,
        "avg_SysQty": 17,
        "stDev_SysQty": 143
    },
    "B": {
        "avg_rowcount": 10,
        "stDev_rowcount": 2,
        "avg_CntQty": 27,
        "stDev_CntQty": 193,
        "avg_SysQty": 27,
        "stDev_SysQty": 309
    },
    "C": {
        "avg_rowcount": 3,
        "stDev_rowcount": 2,
        "avg_CntQty": 50,
        "stDev_CntQty": 310,
        "avg_SysQty": 51,
        "stDev_SysQty": 3090
    },
    "D": {
        "avg_rowcount": 36,
        "stDev_rowcount": 99,
        "avg_CntQty": 22,
        "stDev_CntQty": 31,
        "avg_SysQty": 21,
        "stDev_SysQty": 31
    },
    "E": {
        "avg_rowcount": 35,
        "stDev_rowcount": 120,
        "avg_CntQty": 40,
        "stDev_CntQty": 116,
        "avg_SysQty": 40,
        "stDev_SysQty": 116
    },
    "F": {
        "avg_rowcount": 4,
        "stDev_rowcount": 2,
        "avg_CntQty": 89,
        "stDev_CntQty": 3352,
        "avg_SysQty": 88,
        "stDev_SysQty": 3359
    },
    "G": {
        "avg_rowcount": 2,
        "stDev_rowcount": 2,
        "avg_CntQty": 599,
        "stDev_CntQty": 28430,
        "avg_SysQty": 599,
        "stDev_SysQty": 28430
    }
}
warehouses = pd.DataFrame.from_dict(warehouse_data).T
warehouses.index.name = 'Warehouse_ID'

with_dates = pd.merge(
    left=warehouses.reset_index(),
    right=pd.date_range(start=start_date, end=end_date, name='Date').to_series(),
    how='cross',
)

with_dates['n_rows'] = np.clip(
    a=np.round(
        rand.normal(
            loc=with_dates.avg_rowcount,
            scale=with_dates.stDev_rowcount,
        ),
    ).astype(int),
    a_min=1, a_max=2*with_dates.avg_rowcount - 1,
)

filled = with_dates.loc[with_dates.index.repeat(with_dates.n_rows)]

output = filled[['Warehouse_ID', 'Date']].copy()
output['CntQty'] = np.clip(
    rand.normal(
        loc=filled.avg_CntQty,
        scale=filled.stDev_CntQty,
        size=len(filled),
    ),
    a_min=0, a_max=2*filled.avg_CntQty,
)
output['SysQty'] = np.clip(
    rand.normal(
        loc=filled.avg_SysQty,
        scale=filled.stDev_SysQty,
        size=len(filled),
    ),
    a_min=0, a_max=2*filled.avg_SysQty,
)

for warehouse_id, group in output.groupby('Warehouse_ID'):
    group.to_csv(
        path_or_buf=warehouse_id + 'TaskDetails.csv',
        index=False,
    )

Scaled uniform

Generate your randoms like this:

def uniform_with_std(mean: np.ndarray, std: np.ndarray, min: float) -> np.ndarray:
    """
    If the standard deviation is small enough, use a random distribution with bounds that will
    produce that standard deviation. Otherwise, ignore it so that we can guarantee nothing
    will be returned below the minimum.
    """
    offset = np.sqrt(3) * std
    low = np.clip(a=mean - offset, a_min=min, a_max=None)
    high = 2*mean - low
    return rand.uniform(low=low, high=high, size=mean.size)

This ends up "working" only for your n_rows and ignoring your standard deviations for everything else because they're too high.

Lognorm

Generate your randoms via

dist = rand.lognormal(
    mean=np.log(u*u / np.sqrt(u*u + s*s)),
    sigma=np.sqrt(np.log(1 + s*s/u/u)),
    size=...,
)

This is guaranteed to generate non-negative numbers and respect both your mean (u) and standard deviation (s), but is not normally distributed.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.