If you need further clarification on this script, specific functionalities, or assistance with another aspect, feel free to ask!


in this

-- coding: utf-8 --
"""
Created on Fri Apr 21 22:35:22 2023

@author: akovvuri
"""

import pandas as pd
import numpy as np
#import datetime
import os
import pickle

print("\nSource brokerStandardization.py", dot20)
print(
"\n Purpose: Sub functions used to process and standardize all broker records"
)

badBrokTerms = getVariousTerms(var="BROKER_NAME")+ missingTerms
badBrokTerms=list(dict.fromkeys(badBrokTerms))

broks_set_blank = ["Missing", imputed_term, "Missing Broker", "Not Available"]

broks_set_other = np.unique(
[
"Robelynn Abadie",
"Child Health Corp",
"Child Health Corporation",
"Connecticut General Life",
"Warner",
"Delta Dental",
"Wanada Business Services Corporatio",
"Phra Global Business",
"Various Attached",
"Wanada Business Services Corporation",
"Various Attached",
"Colonial",
]
)

def brokers_manualClean(inFile="brokers.all.yrs.noClean", outFile="brokers.all.yrs.BasicClean", thisDir, REGEN_NAME_CLEAN=True, REGEN_ADDR_CLEAN=True):
def brokers_manualClean(
thisDir,
inFile="brokers.all.yrs.noClean",
outFile="brokers.all.yrs.BasicClean",
REGEN_NAME_CLEAN=True,
REGEN_ADDR_CLEAN=True,
):
print("\n\nStart brokers_manualClean .Load ", inFile, " ")
init = pd.read_pickle(f"{thisDir}{inFile}.pkl")
print(" obs:", init.shape[0])


uniqueVars = [
    "BROKER_INITIAL",
    "BROKER_ZIP_INITIAL",
    "BROKER_ADDR_INITIAL",
    "BROKER_STATE_INITIAL",
    "BROKER_CITY",
    "BROKER_EIN",
]

init.BROKER_ZIP_INITIAL = np.where(
    init.BROKER_ZIP_INITIAL == "nan", "", init.BROKER_ZIP_INITIAL
)
init=init[init.BROKER_INITIAL.isin(['BROOKE AGENCY SERVICES COMPANY LLC',

#                          'BROOKE AGENCY SERVICES COMPANY, LLC'])]
# Remove duplicates
dt = init[uniqueVars].drop_duplicates().reset_index(drop=True)
#.columns

print(" and obs n init after reduce:", dt.shape[0])
obs = dt.shape[0]

# Add a new column ZIP
dt["Rules"] = ""
dt["BROKER_ZIPCODE"] = dt["BROKER_ZIP_INITIAL"].str.zfill(5)
dt = fix_zipcode_vars(dt=dt, v="BROKER_ZIPCODE", stateVar="BROKER_STATE_INITIAL")
dt.isnull().sum()

# Merge with Zipcode to adjust state
map = zipToState.copy()

print("\nMerge with ZiptoStte to adjust state ")
map.rename(
    columns={"ZIPCODE": "BROKER_ZIPCODE", "STATE": "BROKER_STATE"}, inplace=True
)

dt = CustomJoin(
    leftDt=dt, rightDt=map, byVars=["BROKER_ZIPCODE"], txt="zipToState @38"
)

# filling missing values with Broker_state
dt["BROKER_STATE"] = np.where(
    dt["BROKER_STATE"] == "", dt["BROKER_STATE_INITIAL"], dt["BROKER_STATE"]
)

# clean up the broker names & addresses
nameCleanLeftJoin = "brokerName_initialToClean"
addressCleanLeftJoin = "brokerAddress_initialToClean"

# random one-off enforcements
if REGEN_NAME_CLEAN:
    # Generate cleaned up names
    map = dt[["BROKER_INITIAL"]].copy()
    map["BROKER_NAME"] = map["BROKER_INITIAL"]

    map = cleanAndTrimNames(
        dt=map,
        var="BROKER_NAME",
        MANUAL_ADDS=False,
        HUMAN_NAMES=True,
        CAR_OR_BROK=True,
    )
    map = map.loc[
        map["BROKER_NAME"] != map["BROKER_INITIAL"],
        ["BROKER_INITIAL", "BROKER_NAME"],
    ].drop_duplicates()
    pickle.dump(
        map, open(os.path.join(thisDir, "temp", nameCleanLeftJoin + ".pkl"), "wb")
    )
    print(
        "\nSaved BROKER_INITIAL to clean BROKER_NAME map ",
        nameCleanLeftJoin,
        " in/temp @86 ",
        len(map),
    )
else:
    # load the exisitng cleaned up names
    map = pickle.load(
        open(os.path.join(thisDir, "temp", nameCleanLeftJoin + ".pkl"), "rb")
    )

# Join the cleaned up names & data table
dt = CustomJoin(
    leftDt=dt, rightDt=map, byVars=["BROKER_INITIAL"], txt="nameClean @57"
)
dt["BROKER_NAME"] = np.where(
    dt["BROKER_NAME"] == "", dt["BROKER_INITIAL"], dt["BROKER_NAME"]
)

del map
del init

# random one-off enforcements
dt["BROKER_NAME"] = np.where(
    dt["BROKER_INITIAL"].str.contains("DIVERSIFIED INSURANCE BEN"),
    "DIVERSIFIEDINSBEN",
    dt["BROKER_NAME"],
)
dt["BROKER_NAME"] = np.where(
    dt["BROKER_INITIAL"].str.contains("DIVERSIFIED")
    & dt["BROKER_ADDR_INITIAL"].str.contains("136 TEMP"),
    "DIVERSIFIEDINSBEN",
    dt["BROKER_NAME"],
)

# dt['BROKER_NAME'] = dt.apply(lambda x: 'DIVERSIFIEDINSBEN' if 'DIVERSIFIED INSURANCE BEN' in x['BROKER_INITIAL'] else x['BROKER_NAME'], axis=1)
# dt['BROKER_NAME'] = dt.apply(lambda x: 'DIVERSIFIEDINSBEN' if 'DIVERSIFIED' in x['BROKER_INITIAL'] and '136 TEMP' in x['BROKER_ADDR_INITIAL'] else x['BROKER_NAME'], axis=1)

print(
    f"\nUnique BROKER_ADDRESS values on all {len(dt['BROKER_ADDR_INITIAL'].unique())} out of {len(dt)}"
)

if REGEN_ADDR_CLEAN:
    map = dt[dt["BROKER_ADDR_INITIAL"] != ""].drop_duplicates(
        subset=["BROKER_ADDR_INITIAL"]
    )[["BROKER_ADDR_INITIAL"]]
    map["BROKER_ADDRESS"] = map["BROKER_ADDR_INITIAL"]
    map = cleanAddressField(dt=map, var="BROKER_ADDRESS")
    map = map.drop_duplicates(subset=["BROKER_ADDRESS", "BROKER_ADDR_INITIAL"])[
        ["BROKER_ADDRESS", "BROKER_ADDR_INITIAL"]
    ]
    map.to_pickle(f"{thisDir}/temp/{addressCleanLeftJoin}.pkl")
else:
    map = pd.read_pickle(f"{thisDir}/temp/{addressCleanLeftJoin}.pkl")

dt = CustomJoin(
    leftDt=dt,
    rightDt=map,
    byVars=["BROKER_ADDR_INITIAL"],
    txt="addressCleanLeftJoin",
)
dt["BROKER_ADDRESS"] = np.where(
    dt["BROKER_ADDRESS"] == "", dt["BROKER_ADDR_INITIAL"], dt["BROKER_ADDRESS"]
)
print(f"\nUnique address after leftJoin @117 {len(dt['BROKER_ADDRESS'].unique())}")
list_to = (
    [
        "",
        " ",
    ]
    + [str(i) for i in range(1, 11)]
    + missingTerms
)
len(list_to)

dt["FL"] = np.where(dt["BROKER_ADDRESS"].isin(list_to), 1, 0)

print(
    "\nAddresses to revert bc reduced too far ",
    dt["FL"].mean(),
    " out of ",
    dt.shape[0],
)
dt["BROKER_ADDRESS"] = np.where(
    dt["FL"] == 1, dt["BROKER_ADDR_INITIAL"], dt["BROKER_ADDRESS"]
)

# TESTING
na = dt.columns[dt.isna().any()].tolist()
if len(na) > 0:
    stopped = dt
    raise Exception(f"{na} - NAs @88")

# Address STFF

dt["BROKER_ADDRESS"] = dt["BROKER_ADDRESS"].str.replace("  ", " ")
dt["BROKER_ADDRESS"] = dt["BROKER_ADDRESS"].str.strip()

pct = lasj_round(100 * len(dt[dt["BROKER_ADDRESS"] != ""]) / len(dt))
print(
    f"\nObs @112 {len(dt)} + Unique BROKER_ADDRESS b4 bootstrapFuzzy {len(dt['BROKER_ADDRESS'].unique())} % non-missing {pct}"
)

dt = cleanupJunk(dt=dt, otherJunk=["NCHAR"])

dt = dt.rename(columns={"BROKER_ADDRESS": "ADDRESS_INITIAL"})

ag = dt[
    (dt["BROKER_ZIPCODE"] != "")
    & (dt["ADDRESS_INITIAL"] != "")
    & (~dt["BROKER_NAME"].isin(missingTerms))
]
ag = (
    ag.groupby(["BROKER_NAME", "ADDRESS_INITIAL", "BROKER_ZIPCODE", "BROKER_CITY"])
    .agg({"BROKER_EIN": "count"})
    .reset_index()
)
ag = ag.rename(columns={"BROKER_EIN": "GroupsXBrokXAdd"})

map = CREATEbootstrapFuzzy_brokerAddress(
    dt=ag.copy(), outFile="brokerAddressBootstrapFuzzy"
)

# View(map)
byVars = ["BROKER_ZIPCODE", "BROKER_NAME", "ADDRESS_INITIAL"]
dt = CustomJoin(
    leftDt=dt, rightDt=map, byVars=byVars, txt="bootstrap fuzzy on broker addr @117"
)
# dt['BROKER_ADDRESS'] = dt.apply(lambda x: x['ADDRESS_INITIAL'] if x['BROKER_ADDRESS'] == '' else x['BROKER_ADDRESS'], axis=1)
dt["BROKER_ADDRESS"] = np.where(
    dt["BROKER_ADDRESS"] == "", dt["ADDRESS_INITIAL"], dt["BROKER_ADDRESS"]
)

pct = lasj_round(100 * len(dt[dt["BROKER_ADDRESS"] != ""]) / len(dt))
print(
    f"\nUnique BROKER_ADDRESS @128 after bootstrapFuzzy {len(dt['BROKER_ADDRESS'].unique())} % non-missing {pct}"
)

dt = cleanupJunk(dt=dt, otherJunk=["ADDRESS_INITIAL"])

if "Run" not in dt.columns:
    dt["Run"] = f"CleanBrokers-{mmdd}"
else:
    dt["Run"] = dt.apply(lambda x: f"{x['Run']} CleanBrokers-{mmdd}", axis=1)

dt.to_pickle(f"{thisDir}{outFile}.pkl")

print(
    f"\nEND manualClean. Obs in {outFile} {len(dt)} unique BROKER_NAME {len(dt['BROKER_NAME'].unique())} & BROKER_ADDRESS: {len(dt['BROKER_ADDRESS'].unique())} columns: {dt.columns.tolist()}{dot20end}\n"
)
end compileBrokers_clean
def standardizeBROKER(
thisDir,
inFile="brokers.all.yrs.Cleaned",
outFile="brokers.all.yrs.Standardized.B4Address",
):
global broker_strings
mod_start = datetime.now()
print(f"\nStart standardizeBROKER. Loading {inFile}")
dt = pd.read_pickle(f"{thisDir}{inFile}.pkl")
print(f"obs {len(dt)}")


dt = cleanupJunk(dt=dt)
dt.BROKER_ZIP_INITIAL=np.where(dt.BROKER_ZIP_INITIAL.isin(["","0","00"]),
                               "",
                               dt.BROKER_ZIP_INITIAL.str.zfill(5))
 
dt.BROKER_ZIPCODE=np.where(dt.BROKER_ZIPCODE.isin(["","0","00"]),
                               "",
                               dt.BROKER_ZIPCODE.str.zfill(5))

# Stringdetectfixed
dt["BROKER_NAME"] = np.where(
    dt["BROKER_INITIAL"].str.contains("PARADIGM GROUP"),
    "PARADIGMGRP",
    dt["BROKER_NAME"],
)
dt["BROKER_NAME"] = np.where(
    dt["BROKER_INITIAL"].str.contains("PARADIGM CONSULT"),
    "PARADIGMCONSULT",
    dt["BROKER_NAME"],
)
dt["BROKER_NAME"] = np.where(
    dt["BROKER_INITIAL"].str.contains("PARADIGM AGENC"),
    "PARADIGMAGENCY",
    dt["BROKER_NAME"],
)
dt["BROKER_NAME"] = np.where(
    dt["BROKER_INITIAL"].str.contains("PARADIGM INSUR"),
    "PARADIGMAGENCY",
    dt["BROKER_NAME"],
)
dt["BROKER_NAME"] = np.where(
    dt["BROKER_INITIAL"].str.contains("CHAPMAN GROUP"),
    "CHAPMANGRP",
    dt["BROKER_NAME"],
)
dt["BROKER_NAME"] = np.where(
    dt["BROKER_INITIAL"].str.contains("360"), "360 BENEFITS", dt["BROKER_NAME"]
)
dt["BROKER_NAME"] = np.where(
    dt["BROKER_INITIAL"].str.contains("BURNHAM FIN"),
    "BURNHAMFIN",
    dt["BROKER_NAME"],
)

dt["BROKER_NAME"] = np.where(
    dt["BROKER_INITIAL"].str.contains("BURNHAM BENEFITS INSURANCE SER"),
    "BURNHAMBENINS",
    dt["BROKER_NAME"],
)
dt["BROKER_NAME"] = np.where(
    dt["BROKER_INITIAL"].str.contains("BURNHAM BEN"),
    "BURNHAMBEN",
    dt["BROKER_NAME"],
)
mitt = [""] + revertToInit + missingTerms
dt["BROKER_NAME"] = np.where(
    dt["BROKER_NAME"].isin(mitt), dt["BROKER_INITIAL"], dt["BROKER_NAME"]
)
dt[dt.BROKER_INITIAL=='EMERSON REID LLC'].BROKER_NAME

Rule = "ReplaceHyphenCommaLLCWithSpace (again)"
strFrom = [",", "-", "&", "LLC", "  b", "  ", "  "]
strTo = [" "] * len(strFrom)

dt = executeCleaning(
    dt=dt.copy(), var="BROKER_NAME", stringsFrom=strFrom, stringsTo=strTo, Rule=Rule
)
print(f"\nEnd {Rule} @181. Impacted .{dt['FL'].sum()}")

Rule = "HarmonizeBENEFITtoBENEFITS (again)"
strFrom = ["BENEFIT", "BENEFITSS"]
strTo = ["BENEFITS"] * len(strFrom)
dt = executeCleaning(
    dt=dt.copy(), var="BROKER_NAME", stringsFrom=strFrom, stringsTo=strTo, Rule=Rule
)
print(f"\nEnd {Rule}. Impacted .{dt['FL'].sum()}")

dt = removeSpaces_LeadingTrailing(dt=dt.copy(), var="BROKER_NAME")
print(f"\nObs @190 {len(dt)} unique broker names {len(dt['BROKER_NAME'].unique())}")
dt["BROKER_ADDRESS"] = dt["BROKER_ADDRESS"].str.replace("  ", " ")
dt["BROKER_ADDRESS"] = dt["BROKER_ADDRESS"].str.strip()

badNameCheck(dtCheck=dt, var="BROKER_NAME", txt="@201")

dt = dt.rename(columns={"BROKER_NAME": "BROKER_NAME_ORIG"})

map = broker_exact[["BROKER_STRING", "BROKER"]]

sub = dt[["BROKER_EIN", "BROKER_NAME_ORIG"]].drop_duplicates()
sub["BROKER_BY_EIN"] = sub["BROKER_EIN"].map(
    dict(zip(map["BROKER_STRING"], map["BROKER"]))
)


sub["BROKER_BY_EIN"] = sub["BROKER_BY_EIN"].fillna("")
print(sub["BROKER_BY_EIN"].value_counts())
sub["BROKER_NAME"] = sub["BROKER_NAME_ORIG"].map(
    dict(zip(map["BROKER_STRING"], map["BROKER"]))
)
sub["BROKER_NAME"] = sub["BROKER_NAME"].fillna("")
print(sub["BROKER_BY_EIN"].value_counts())
sub["BROKER_STRING"] = sub.apply(
    lambda x: "" if x["BROKER_NAME"] == "" else x["BROKER_NAME"], axis=1
)
sub["BROKER_STRING"].value_counts(dropna=False)
sub["METHOD"] = sub.apply(
    lambda x: "EXACT" if x["BROKER_STRING"] != "" else "", axis=1
)
sub["METHOD"].value_counts(dropna=False)

na = sub.columns[sub.isna().any()].tolist()
if len(na) > 0:
    print(na)
    raise Exception("-NAs @250")
if "qcLogBrok" in globals():
    del globals()["qcLogBrok"]

print(f"\nStart broker_string search method Obs in sub @ 231 {len(sub)}")
broker_strings = broker_strings.sort_values(

#    by=["BROKER", "ORDER", "BROKER_STRING"], ascending=[True, True, True]
#)
broker_strings=broker_strings.reset_index(drop=True)

sub=sub.reset_index(drop=True)
tempLog = None
for i in range(len(broker_strings)):
    broker_strings["BROKER"]
    string = broker_strings.loc[i, "BROKER_STRING"]
    broker = broker_strings.loc[i, "BROKER"]
    sub["MATCH"] = np.where(
        (sub["BROKER_STRING"] == "")
        & (sub["BROKER_NAME_ORIG"].str.contains(string, na=False)),
        1,
        0,
    )
    print(broker,string)
    Caught = sub["MATCH"].sum()
    if Caught==1:
        print(broker,string)
    Single = ""
    if Caught == 1:
        te = sub[sub["MATCH"] == 1]
        Single = te.iloc[0]["BROKER_NAME_ORIG"]
    
    add = pd.DataFrame(
        [[string, broker, Caught, Single]],
        columns=["string", "broker", "Caught", "Single"],
    )

    if i == 0:
        tempLog = add
    else:
        tempLog = pd.concat([tempLog, add], ignore_index=True)

    sub["BROKER_NAME"] = np.where(sub["MATCH"] == 1, broker, sub["BROKER_NAME"])
    sub["BROKER_STRING"] = np.where(sub["MATCH"] == 1, string, sub["BROKER_STRING"])

qcLogBrok = tempLog
# qcLogBrok.head()
sub.to_pickle(f"{thisDir}subPostString.pkl")
print(f"\nSaved interim as thisDir/subPostString @245 {len(sub)}")

sub.loc[:, "METHOD"] = np.where(
    (sub["METHOD"] == "") & (sub["BROKER_STRING"] != ""), "STRING", sub["METHOD"]
)
sub.METHOD.value_counts(dropna=False)
sub.loc[:, "FL"] = np.where(
    (sub["BROKER_STRING"] == "") & (sub["BROKER_BY_EIN"] != ""), 1, 0
)

print(
    "\nEnd search string. residual obs can std by EIN",
    sub["FL"].sum(),
    "of",
    len(sub),
)
sub.loc[:, "BROKER_STRING"] = np.where(
    sub["FL"] == 1, sub["BROKER_EIN"], sub["BROKER_STRING"]
)
sub.loc[:, "BROKER_NAME"] = np.where(
    sub["FL"] == 1, sub["BROKER_BY_EIN"], sub["BROKER_NAME"]
)
sub.loc[:, "METHOD"] = np.where(sub["FL"] == 1, "EIN", sub["METHOD"])

std = sub[sub["BROKER_STRING"] != ""]
std = std[
    ["BROKER_NAME", "BROKER_STRING", "METHOD", "BROKER_NAME_ORIG", "BROKER_EIN"]
].drop_duplicates()
len(std)
print(
    f"\nNames on std after partial string @232 {len(std)} out of {len(sub)} on sub"
)
print(set(dt.columns).intersection(std.columns))
# dt = pd.merge(dt, std, on=['BROKER_NAME_ORIG', 'BROKER_EIN'], how='left', suffixes=('', '_y'))
dt = CustomJoin(
    leftDt=dt,
    rightDt=std,
    byVars=["BROKER_NAME_ORIG", "BROKER_EIN"],
    txt="standardized broker names @244",
)
dt["STAND"] = dt.apply(lambda x: 1 if x["BROKER_STRING"] != "" else 0, axis=1)
print(f"\nTotal standardized {dt['STAND'].mean()} and Dist of method")
print(dt["METHOD"].value_counts())
badNameCheck(dtCheck=dt, var="BROKER_NAME", txt="@263")
dt = cleanupJunk(dt=dt, otherJunk=["STAND"])
dt["Run"] = dt.apply(lambda x: f"{x['Run']} standBROKER{mmdd}", axis=1)
dt.to_pickle(f"{thisDir}{outFile}.pkl")
print(f"\nObs saved in {outFile} {len(dt)}")
end StandardizeBROKER functon
def standardizeBROKERwithADDRESS(
thisDir,
baseInFile,
inFile="brokers.all.yrs.Standardized.B4Address",
outFile="brokers.StandardizedStaging",
addressMapFile="BrokerAddressZipcodeMap",
PRINT_QC=True,
):
print(f"\nStart standardizebrokerw/address. Load {inFile}")


dt = pd.read_pickle(f"{thisDir}{inFile}.pkl")

print(f"obs {len(dt)}")



createBrokerAddressMap(dt=dt, thisDir=brokerDir, outFile=addressMapFile)

map = pd.read_pickle(f"{thisDir}{addressMapFile}.pkl")
print(f"\nObs {addressMapFile} {len(map)}")
dt = dt.rename(columns={"BROKER_NAME": "BROKER", "BROKER_NAME_ORIG": "BROKER_ORIG"})

dt["FL"] = np.where(dt["BROKER"].isin(broker_exact["BROKER"].tolist()), 1, 0)

dt["BROKER_ZIP_ORIG_DELME"] = dt["BROKER_ZIPCODE"]
dt["BROKER_ZIPCODE"] = dt["BROKER_ZIPCODE"].str[:5]

badNameCheck(dtCheck=dt, var="BROKER", txt="@500")
map = map[
    (map["BROKER_NAME"].isin(broker_exact["BROKER"]))
    & (~map["BROKER_NAME"].isin(missingTerms))
]
print(
    f"\nInitial addresses X states mapped to standardized brokers @408 {len(map)}"
)
map["GroupsXAdd"] = map.groupby(
    ["BROKER_STATE", "BROKER_ZIPCODE", "BROKER_ADDRESS"]
)["Groups"].transform("sum")
map["PctAddWithBrok"] = lasj_round(map["Groups"] / map["GroupsXAdd"], 2)
map["RATIO"] = lasj_round(map["Other_STAND"] / map["Groups"], 2)
map = map[
    (map["PctAddWithBrok"] > 0.5) & (map["Groups"] >= 9) & (map["RATIO"] < 0.25)
]
print(f"Obs w/50%+ of the groups there @413 {len(map)}")

map = map[
    ["BROKER_NAME", "BROKER_ZIPCODE", "BROKER_ADDRESS", "BROKER_STATE"]
].drop_duplicates()
map = map.rename(columns={"BROKER_NAME": "XWALK"})

dt = pd.merge(
    dt, map, on=["BROKER_ZIPCODE", "BROKER_ADDRESS", "BROKER_STATE"],
    how="left"
)

dt["XWALK"] = dt["XWALK"].fillna("")

dt["FL"] = np.where((dt["BROKER"].isin(broker_exact["BROKER"].tolist())==False) & 
                    (dt["XWALK"] != ""), 1, 0)

print(" Obs to stand by address @430", dt["FL"].sum())

dt["BROKER"] = np.where(dt["FL"] == 1, dt["XWALK"], dt["BROKER"])
dt["BROKER_STRING"] = np.where(
    dt["FL"] == 1, dt["BROKER_ADDRESS"], dt["BROKER_STRING"]
)
dt["BROKER"] = np.where(
    (dt["BROKER"].isin(missingTerms)) & ~(dt["BROKER_ORIG"].isin(missingTerms)),
    dt["BROKER_ORIG"],
    dt["BROKER"],
)
if "METHOD" in dt.columns:
    dt.loc[:, "METHOD"] = np.where(
        (dt["FL"] == 1) & (dt["METHOD"] == ""), "ADDRESS", dt["METHOD"]
    )
    print(dt["METHOD"].value_counts())

rule = "\nManually add COLONIAL LIFE if in the address field ."
dt["FL"] = dt.apply(
    lambda x: 1
    if "COLONIAL" in x["BROKER_ADDRESS"] and x["BROKER"] != "COLONIAL LIFE (UNUM)"
    else 0,
    axis=1,
)
dt["BROKER"] = dt.apply(
    lambda x: "COLONIAL LIFE (UNUM)" if x["FL"] == 1 else x["BROKER"], axis=1
)
dt["BROKER_STRING"] = dt.apply(
    lambda x: "NAME IN ADDR" if x["FL"] == 1 else x["BROKER_STRING"], axis=1
)
print(f"{rule}{dt['FL'].sum()}")

rule = "\nManually Add AFLAC if in th ADDRESS field "
dt["FL"] = dt.apply(
    lambda x: 1 if "AFLAC" in x["BROKER_ADDRESS"] and x["BROKER"] != "AFLAC" else 0,
    axis=1,
)
dt["BROKER_STRING"] = dt.apply(
    lambda x: "NAME IN ADDR" if x["FL"] == 1 else x["BROKER_STRING"], axis=1
)
dt["BROKER"] = dt.apply(lambda x: "AFLAC" if x["FL"] == 1 else x["BROKER"], axis=1)
print(f"{rule}{dt['FL'].sum()}")

rule = "\nManually Add USI if WELLS FARGO in address field "
dt["FL"] = dt.apply(
    lambda x: 1
    if "WELLS FAR" in x["BROKER_ADDRESS"] and x["BROKER"] != "USI"
    else 0,
    axis=1,
)
dt["BROKER_STRING"] = dt.apply(
    lambda x: "WELLS FARGO" if x["FL"] == 1 else x["BROKER_STRING"], axis=1
)
dt["BROKER"] = dt.apply(lambda x: "USI" if x["FL"] == 1 else x["BROKER"], axis=1)
print(f"{rule}{dt['FL'].sum()}")

rule = "\nManually change for BENEFIT SERVICES GROUP in WI"
dt["FL"] = dt.apply(
    lambda x: 1
    if (
        "BENEFITS SERVICE" in x["BROKER_STRING"]
        or "THE BENEFITS GROU" in x["BROKER_STRING"]
    )
    and x["BROKER_STATE"] == "WI"
    else 0,
    axis=1,
)
dt["BROKER_STRING"] = dt.apply(
    lambda x: "BENEFITS SERVICES GROUP" if x["FL"] == 1 else x["BROKER_STRING"],
    axis=1,
)
dt["BROKER"] = dt.apply(
    lambda x: "BENEFITS SERVICES GROUP" if x["FL"] == 1 else x["BROKER"], axis=1
)
print(f"{rule}{dt['FL'].sum()}")

toName = "BURNHAM BENEFITS INSURANCE SERVICES"
rule = f"\nOne off for {toName}"
dt["FL"] = dt.apply(
    lambda x: 1 if "BURNHAM BENEFITS INSURANCE SER" in x["BROKER_INITIAL"] else 0,
    axis=1,
)
dt["FL"] = dt.apply(
    lambda x: 1
    if "BURNHAM BEN" in x["BROKER_INITIAL"]
    and x["BROKER_ZIPCODE"] in [92612, "92612"]
    else x["FL"],
    axis=1,
)
dt["BROKER"] = dt.apply(lambda x: toName if x["FL"] == 1 else x["BROKER"], axis=1)
dt["BROKER_STRING"] = dt.apply(
    lambda x: toName if x["FL"] == 1 else x["BROKER_STRING"], axis=1
)

DOTHIS = False
if DOTHIS:
    rule = "\nFlag as direct if entered carrier name VS broker names ."
    carrierNotBroker = [
        "DELTA DENT",
        "ANTHEM",
        "BLUE CROSS",
        "BCBS",
        "CIGNA",
        "HIGHMARK",
        "HCSC",
        "HUMANA",
        "SUNLIFE",
        "METLIFE",
    ]
    dt["FL"] = 0
    for string in carrierNotBroker:
        dt["FL"] = dt.apply(
            lambda x: 1 if string in x["BROKER_ORIG"].upper() else x["FL"], axis=1
        )
    majorCarriers = [x.upper() for x in majorMedical + majorDental + majorLifeDisab]
    dt["FL"] = dt.apply(
        lambda x: 1 if x["BROKER"] in majorCarriers else x["FL"], axis=1
    )
    print(f"{rule} {dt['FL'].sum()}")
    dt["BROKER"] = np.where(dt["FL"] == 1, "DIRECT", dt["BROKER"])

rule = "\nDrop if broker name is missing/varoius/etc "
dt["DROP"] = dt.apply(lambda x: 1 if x["BROKER"] in badBrokTerms else 0, axis=1)

print(f"{rule}{dt['DROP'].sum()}")
dt = dt[dt["DROP"] == 0]
dt["STAND"] = dt.apply(
    lambda x: 1 if x["BROKER"] in broker_exact["BROKER"].values else 0, axis=1
)
print(f"std @ 500 {dt['STAND'].mean()}")

dt["BrokerProper"] = dt["BROKER"]
dt = revert_case(dt, varList=["BrokerProper"])
dt = cleanupJunk(dt=dt, otherJunk=["STAND"])
print(dt.columns)

# LEFT JOIN WTH FULL SET OF FIELDS
print(
    f"\nLoad {baseInFile} baseline file initially loaded for left join w/initial fields."
)
map = dt.copy()
dt = pd.read_pickle(f"{thisDir}{baseInFile}.pkl")
print(f"obs {len(dt)}")
byVars = [
    "BROKER_INITIAL",
    "BROKER_ADDR_INITIAL",
    "BROKER_ZIP_INITIAL",
    "BROKER_STATE_INITIAL",
    "BROKER_EIN",
]

int = set(map.columns).intersection(dt.columns)
dif = set(byVars).difference(int)
if len(dif) > 0:
    raise Exception(f"byVars missing @387 {dif}")
dt.BROKER_ZIP_INITIAL=np.where(dt.BROKER_ZIP_INITIAL.isin(["","0","00"]),
                               "",
                               dt.BROKER_ZIP_INITIAL.str.zfill(5))

otherDrop = set(int).difference(byVars)
if len(otherDrop) > 0:
    print(f"\nOther vars to drop from mapping file @388 {otherDrop}")
    map = map.drop(columns=otherDrop)
    map = map.drop_duplicates()
    map=map.sort_values(byVars+['Rules'], ascending=[True,True,True,True,True,False])
    map=map.drop_duplicates(subset=byVars,keep='first')

# dt = pd.merge(dt, map, on=byVars, how='left', suffixes=('', '_y'))
dt = CustomJoin(leftDt=dt, rightDt=map, byVars=byVars)

midFile = "mid_standWithAddress"

print(
    "\nAdd the base filing vars @400 - PN, Part_EOY, welfarePn vs RetrementPn, etc "
)
base = pd.read_pickle(f"{pitAllYearsDir}e5500.all_allYears_essentialVars.pkl")
byVars = ["ACK_ID"]
keepVars = list(
    set(
        byVars
        + [
            "State_HQ",
            "Zipcode",
            "EIN",
            "PN",
            "WelfarePn",
            "RetirementPn",
            "Part_EOY",
        ]
    )
)
base = base[keepVars].drop_duplicates()
dt = CustomJoin(
    leftDt=dt, rightDt=base, byVars=byVars, INNER=True, txt="base filing vars @400"
)

dt["STAND"] = np.where(dt["BROKER"].isin(broker_exact["BROKER"]), 1, 0)
print(f"std @400 {dt['STAND'].mean()}")

rule = (
    "\nDrop if actuary and also there is a broker (if not retirementP) - sc C only"
)
# ct = dt.groupby(["ACK_ID"])["IsBroker"].sum().reset_index(name="CT")
temp_dt = (
    dt.groupby(["ACK_ID", "FORM_ID"], as_index=False)
    .agg(CT=("IsBroker", "sum"))
    .reset_index()
)
dt = pd.merge(dt, temp_dt, on=["ACK_ID", "FORM_ID"])
dt["DROP"] = np.where(
    (dt["Sch"] == "C")
    & (dt["CT"] > 0)
    & (dt["IsActuary"] == 1)
    & (dt["IsBroker"] == 0),
    1,
    0,
)
print(rule, dt["DROP"].sum(), " vs actuary ", dt["IsActuary"].sum())
dt = dt[dt["DROP"] == 0]
dt["IsOther"] = np.where(dt["IsActuary"] == 1, 1, dt["IsOther"])
dt = dt.drop(["CT"], axis=1)
rule = "\nDrop if not a welfare insurance broker ."
temp_dt = (
    dt.groupby(["ACK_ID"], as_index=False).agg(CT=("IsBroker", "sum")).reset_index()
)
dt = pd.merge(dt, temp_dt, on=["ACK_ID"])

dt["DROP"] = np.where(
    (dt["CT"] > 0)
    & (dt["RetirementPn"] == 0)
    & (dt["IsBroker"] == 0)
    & (dt["WelfarePn"] == 1)
    & (dt["ROW_ORDER"] > 1)
    & (
        (dt["IsBrokerNonInsurance"] == 1)
        | (dt["IsMutualFund"] == 1)
        | (dt["IsBank"] == 1)
        | (dt["IsTrust"] == 1)
        | (dt["IsInvAdvisor"] == 1)
    ),
    1,
    0,
)
print(rule, dt["DROP"].sum(), " out of ", len(dt))
dt = dt[dt["DROP"] == 0]

MAYBE_DO_THIS = False
if MAYBE_DO_THIS:
    dt["DROP"] = np.where((dt["IsBroker"] == 0) & (dt["IsOther"] == 0), 1, 0)
    print(
        f" DROP bc nether isBroker or isOther {dt['DROP'].sum()} pct {dt['DROP'].mean()}"
    )
    te = dt[
        (dt["DROP"] == 1)
        & (dt["WelfarePn"] == 1)
        & (dt["IsTPA"] == 0)
        & (dt["STAND"] == 1)
    ]
    dt = dt[dt["DROP"] == 0]

    rule = "\nDrop if wasn't isBroker=1 and is not standardized "
    dt["DROP"] = np.where((dt["isBroker"] == 0) & (dt["STAND"] == 0), 1, 0)
    print(rule, dt["DROP"].sum())
    dt = dt[dt["DROP"] == 0]

dt["Run"] = dt["Run"] + " StdWithAddress " + mmdd
print("\nStandardized ", dt["STAND"].mean())
dt = cleanupJunk(dt=dt, otherJunk=["XWALK", "STAND", "BROKER_ZIP_ORIG_DELME"])
dt.to_pickle(thisDir + outFile + ".pkl")
print(f"  and obs saved in {outFile} {len(dt)} {dot20end}")
end standardizeBROKERwithADDRESS
def createBrokerAddressMap(dt, thisDir, outFile="BrokerAddressZipcodeMap"):
global carrier_exact
global brokAddressMap

dt = pd.read_pickle(f"{thisDir}{inFile}.pkl")

print(
    f"\nStart fn to create {outFile}. Obs init {len(dt)} unique broker names {len(dt['BROKER_NAME'].unique())}"
)
carrier_exact = carrier_exact.rename(columns={"Carrier": "CARRIER"})

notAddress = list(
    set(
        ["", "ADMIN", "29", "AON CONSULTING", "UNKNOWN"]
        + broker_exact["BROKER_STRING"].tolist()
        + carrier_exact["Carrier"].tolist()
    )
)

len(notAddress)

rule = "Harmonize zipcodes if only the last digit is off  implies a filing error ."
dt["KEY"] = dt["BROKER_ZIPCODE"].str[:4]
ag = (
    dt.groupby(["BROKER_NAME", "BROKER_ZIPCODE", "KEY", "BROKER_STATE"])
    .agg({"METHOD": "count"})
    .reset_index()
)
ag = ag.rename(columns={"METHOD": "GroupsXZip"})
ag["GroupsXKey"] = ag.groupby(["BROKER_NAME", "KEY", "BROKER_STATE"])[
    "GroupsXZip"
].transform("sum")

ag = ag[
    ~(ag["BROKER_NAME"].isin(badBrokTerms))
    & (ag["GroupsXZip"] < ag["GroupsXKey"])
    & (ag["BROKER_ZIPCODE"] != "")
]


ag["IN"] = np.where(ag["BROKER_ZIPCODE"].isin(zipToState["ZIPCODE"]), 1, 0)
#can discuss here if we need 
ag.IN.value_counts(dropna=False)
ag = ag.sort_values(
    ["BROKER_NAME", "KEY", "IN", "GroupsXZip"], 
    ascending=[True, True, False, False]
)
ag["IND"] = (
    ag.groupby(["BROKER_NAME", "KEY", "BROKER_STATE"])["GroupsXZip"].cumcount() + 1
)


map = ag[ag["IND"] == 1][
    ["BROKER_ZIPCODE", "KEY", "BROKER_NAME", "BROKER_STATE"]
].reset_index(drop=True)
map = map.rename(columns={"BROKER_ZIPCODE": "XWALK"})
dt = pd.merge(dt, map, on=["BROKER_NAME", "KEY", "BROKER_STATE"], how="left")
dt["XWALK"] = np.where(dt["XWALK"].isna(), "", dt["XWALK"])
dt["FL"] = np.where(
    (dt["XWALK"] != "") & (dt["XWALK"] != dt["BROKER_ZIPCODE"]), 1, 0
)
print("\nEnd ", rule, dt["FL"].mean())
dt["BROKER_ZIPCODE"] = np.where(dt["FL"] == 1, dt["XWALK"], dt["BROKER_ZIPCODE"])
dt["Rules"] = np.where(
    dt["FL"] == 0, dt["Rules"], dt["Rules"] + "ChangeLastZipDigit; "
)
rule = (
    "Harmonize zipcodes if BROKER_NAME and ADDRESS are same but zip is different."
)

byVars = ["BROKER_NAME", "BROKER_ADDRESS", "BROKER_STATE"]

ag = (
    dt.groupby(["BROKER_ZIPCODE"] + byVars)["METHOD"]
    .agg(GroupsXZip="size")
    .reset_index()
)
ag["GroupsXKey"] = ag.groupby(byVars)["GroupsXZip"].transform("sum")
ag = ag[
    (~ag["BROKER_NAME"].isin(badBrokTerms))
    & (ag["GroupsXZip"] < ag["GroupsXKey"])
    & (ag["BROKER_ZIPCODE"] != "")
    & (ag["BROKER_ADDRESS"] != "")
]
ag.columns
ag["IN"] = np.where(ag.BROKER_ZIPCODE.isin(zipuq.ZIPCODE.tolist()), 1, 0)
print(ag.IN.mean())
ag = ag.sort_values(
    by=["BROKER_NAME", "BROKER_ADDRESS", "IN", "GroupsXZip"],
    ascending=[True, True, False, False],
)
ag["IND"] = ag.groupby(byVars).cumcount() + 1
map = ag[ag["IND"] == 1][byVars + ["BROKER_ZIPCODE"]].copy()
map = map.drop_duplicates(subset=byVars, keep="first")
map = map.rename(columns={"BROKER_ZIPCODE": "XWALK"})
map.columns
try:
    dt = dt.drop(["XWALK"], axis=1)
except:
    print("xwalk not der")
dt = dt.merge(map, on=byVars, how="left")
dt["XWALK"] = dt["XWALK"].fillna("")
dt["FL"] = np.where((dt["XWALK"] != "") & (dt["XWALK"] != dt["BROKER_ZIPCODE"]),1,0)
print("\nEnd", rule, dt["FL"].mean())
dt["BROKER_ZIPCODE"] = np.where(dt["FL"] == 1, dt["XWALK"], dt["BROKER_ZIPCODE"])
dt["Rules"] = np.where(
    dt["FL"] == 0, dt["Rules"], dt["Rules"] + "HarmonizeZipcodeBrokNameAddr; "
)


rule = "Harmonize zipcodes if NAME_ORIG and ADDRESS are same but zip is differet ."
byVars = ["BROKER_NAME_ORIG", "BROKER_ADDRESS", "BROKER_STATE"]
ag = dt.groupby(by=["BROKER_ZIPCODE"] + byVars, as_index=False).agg(
    {"METHOD": "count"}
)
ag.rename(columns={"METHOD": "GroupsXZip"}, inplace=True)
ag["GroupsXKey"] = ag.groupby(by=byVars)["GroupsXZip"].transform("sum")
ag = ag[
    (~ag["BROKER_NAME_ORIG"].isin(badBrokTerms))
    & (ag["GroupsXZip"] < ag["GroupsXKey"])
    & (ag["BROKER_ZIPCODE"] != "")
    & (ag["BROKER_ADDRESS"] != "")
]
ag["IN"] = np.where(ag["BROKER_ZIPCODE"].isin(zipToState["ZIPCODE"]),1,0)
ag.sort_values(
    by=["BROKER_NAME_ORIG", "BROKER_ADDRESS", "IN", "GroupsXZip"],
    ascending=[True, True, False, False],
    inplace=True,
)
ag["IND"] = ag.groupby(by=byVars).cumcount() + 1
map = ag.loc[ag["IND"] == 1, byVars + ["BROKER_ZIPCODE"]].drop_duplicates()
map.rename(columns={"BROKER_ZIPCODE": "XWALK"}, inplace=True)
try:
    dt.drop(columns=["XWALK"], inplace=True)
except:
    print("not der xwalk")
dt = dt.merge(map, on=byVars, how="left")
dt["XWALK"].fillna("", inplace=True)
dt["FL"] = np.where((dt["XWALK"] != "") & (dt["XWALK"] != dt["BROKER_ZIPCODE"]),1,0)

print(f"\nEnd {rule} {dt['FL'].mean()}")
dt["BROKER_ZIPCODE"] = np.where(dt["FL"] == 1, dt["XWALK"], dt["BROKER_ZIPCODE"])
dt["Rules"] = np.where(
    dt["FL"] == 0, dt["Rules"], dt["Rules"] + "HarmonizeZipcodeBrokOrigAddr;"
)
varTerms = getVariousTerms(var="BROKER_NAME")
dt["DROP"] = np.where(
    (dt["BROKER_NAME_ORIG"].isin(badBrokTerms))
    | (dt["BROKER_ZIPCODE"] == "")
    | (dt["BROKER_ADDRESS"].isin(notAddress))
,1,0)

print(f"\nPct to drop b/c missing or various {dt['DROP'].mean()}")
dropped = dt[dt["DROP"] == 1]
# dropped.to_csv('ce.csv')
dt = dt[dt["DROP"] == 0]
dt["BROKER_NAME"] = np.where(
    dt["BROKER_NAME"] == "", dt["BROKER_NAME_ORIG"], dt["BROKER_NAME"]
)
dt["STAND"] = np.where(dt["BROKER_NAME"].isin(broker_exact["BROKER"]), 1, 0)
print(dt["STAND"].mean())


rule = "\nMap a single broker address to a broker zipcode"
byVars = ["BROKER_NAME", "BROKER_STATE", "BROKER_ZIPCODE"]

NON_LOCAL_STUFF = False
if NON_LOCAL_STUFF:
    dt["NonLocal"] = np.where(dt["State_HQ"] != dt["BROKER_STATE"], 1, 0)
    print("\n% non-local ", dt["NonLocal"].mean())
    ag = (
        dt.groupby(byVars + ["BROKER_ADDRESS"])
        .agg(Groups=("Year", "size"), NonLocalSum=("NonLocal", "sum"))
        .reset_index()
    )
    ag["GroupsZip"] = ag.groupby(byVars)["Groups"].transform("sum")
    ag["NonLocalSum"] = ag.groupby(byVars + ["BROKER_ADDRESS"])[
        "NonLocalSum"
    ].transform("sum")

ag = (
    dt.groupby(byVars + ["BROKER_ADDRESS"])
    .agg(Groups=("METHOD", "size"))
    .reset_index()
)
# changed based on inputs from mckensy team

# ag[,GroupsZip:=sum(Groups),by=byVars]
ag["Groups"] = 1

ag["GroupsZip"] = ag.groupby(byVars)["Groups"].transform("sum")
ag["GroupsZip"].value_counts(dropna=False)
ag = ag.sort_values(
    by=["BROKER_NAME", "BROKER_STATE", "BROKER_ZIPCODE", "Groups"],
    ascending=[True, True, True, False],
)
#ag["IND"] = ag.groupby(byVars)["Groups"].rank(method="first")
ag["IND"]= ag.groupby(byVars)["Groups"].cumcount() + 1
print("\nObs in address map before restrict ", ag.shape[0])
ag = ag[ag["IND"] == 1].copy()
ag["Groups"] = ag["GroupsZip"]
ag.drop(columns=["GroupsZip"], inplace=True)
print("\nObs in address map when reduce to one address ", ag.shape[0])

if NON_LOCAL_STUFF:
    ag["NonLocal"] = lasj_round(ag["NonLocalSum"] / ag["Groups"], 1)
    ag["NonLocal"].value_counts()

ag = cleanupJunk(ag, otherJunk=["NonLocalSum"])
ag = ag[ag["Groups"] > 1]
ag.Groups.value_counts(dropna=False)
print("\nObs in ag when limit to 3+ groups ", ag.shape[0])
ag["SUM"] = ag.groupby(["BROKER_NAME", "BROKER_STATE"])["Groups"].transform("sum")
ag["GroupsXBrok"] = ag.groupby(["BROKER_NAME"])["Groups"].transform("sum")
ag["GroupsXBrokXSt"] = ag.groupby(["BROKER_NAME", "BROKER_STATE"])[
    "Groups"
].transform("sum")
ag["PCT"] = lasj_round(ag["Groups"] / ag["SUM"], 1)
ag["PCT"].value_counts()
ag["MAX"] = ag.groupby(["BROKER_NAME", "BROKER_STATE"])["Groups"].transform("max")
ag["STAND"] = np.where(
    ag["BROKER_NAME"].isin(broker_exact["BROKER"].tolist()), ag["Groups"], 0
)
ag["Other_STAND"] = ag.groupby(
    ["BROKER_STATE", "BROKER_ZIPCODE", "BROKER_ADDRESS"]
)["STAND"].transform("sum")


ag["Other_STAND"] = ag["Other_STAND"] - ag["STAND"]
ag["OtherGroups"] = ag.groupby(
    ["BROKER_STATE", "BROKER_ZIPCODE", "BROKER_ADDRESS"]
)["Groups"].transform("sum")
ag["OtherGroups"] = ag["OtherGroups"] - ag["Groups"]
ag["RATIO"] = np.where(
    ag["Other_STAND"] < 1, 1, lasj_round(ag["Groups"] / ag["Other_STAND"], 1)
)
ag["DROP"] = np.where(
    (ag["PCT"] < 0.4) & (ag["Groups"] < 10) & (ag["MAX"] >= 10), 1, 0
)
ag["DROP"].value_counts(dropna=False)
if NON_LOCAL_STUFF:
    ag["BrokNonLoc"] = ag["NonLocal"].sum()
    ag["DROP"] = np.where(
        (ag["GroupsXBrok"] < 10) & (ag["BrokNonLoc"] < 0.2), 1, ag["DROP"]
    )
    ag["DROP"] = np.where(
        (ag["Groups"] == ag["GroupsXBrok"])
        & (ag["Groups"] < 20)
        & (ag["BrokNonLoc"] < 0.2),
        1,
        ag["DROP"],
    )
else:
    ag["DROP"] = np.where(ag["GroupsXBrok"] < 9, 1, ag["DROP"])
    # ag["DROP"] = ag["DROP"].mean()

ag = ag[ag["DROP"] == 0]
ag = ag.sort_values(
    by=["BROKER_STATE", "BROKER_ZIPCODE", "BROKER_ADDRESS", "Groups"],
    ascending=[True, True, True, False],
)

ag["DROP"] = np.where(
    (ag["RATIO"] < 0.3)
    & (ag["STAND"] == 0)
    & (ag["Groups"] < 15)
    & (ag["Other_STAND"] > 150),
    1,
    0,
)
print("\nPct to drop b/c most likely non yet standardized ", ag["DROP"].mean())
ag = ag[ag["DROP"] == 0]

ag = cleanupJunk(
    dt=ag, otherJunk=["DROP", "MAX", "PCT", "RATIO", "SUM", "BrokNonLoc"]
)

ag = ag.drop_duplicates()



ag["PctBrokAtAdd"] = lasj_round(ag["Groups"] / ag["GroupsXBrok"], 2)
ag["PctStateBrokAtAdd"] = lasj_round(ag["Groups"] / ag["GroupsXBrokXSt"], 2)

ag.to_pickle(thisDir + outFile + ".pkl")
brokAddressMap = ag.copy()
print("\nObs saved in", outFile, ag.shape[0], "to globalEnvt as brokAddressMap")

# END CREATEBROKERADDRESSMP functon
def appendBaseFields(infile="brokers.StandardizedStaging"):
print("\nLoading ", infile, ".pkl")
dt = pd.read_pickle(os.path.join(brokerDir, infile + ".pkl"))
print(dt.shape)
print("\nDone")
dt["BROKER_STATE"] = np.where(
dt["BROKER_STATE"] == "", dt["BROKER_STATE_INITIAL"], dt["BROKER_STATE"]
)


dt["DROP"] = np.where((dt["BROKER"] == "") & (dt["BROKER_INITIAL"] != ""), 1, 0)
print("\nObs missing broker name but hventiall: ", dt[dt["DROP"] == 1].shape[0])
dt = dt[dt['DROP']==0]
dt = cleanupJunk(
    dt=dt,
    otherJunk=[
        "BROKER_STATE_INITIAL",
        "BROKER_ZIP_INITIAL",
        "BROKER_ADDR_INITIAL",
        "Country",
    ],
)

map = pd.read_pickle(pitAllYearsDir + "schA_prepped_allYears.pkl")
print("\nAdd schedule A variables @411   Obs in map ", len(map))
map["KEY"] = map["ACK_ID"].astype(str) + map["FORM_ID"]

keepVars = [
    "KEY",
    "Carrier_Orig",
    "Carrier_EIN",
    "Carrier_NAIC",
    "CTR_NUM",
    "Members",
    "Medical",
    "Life",
    "Dental",
    "Vision",
    "STD",
    "LTD",
    "Stoploss",
    "Drugs",
    "Premiums",
]

map = map[keepVars].drop_duplicates()

schC = dt[dt["Sch"] == "C"]
schA = dt[dt["Sch"] == "A"]
schC["FORM_ID"]=schC["FORM_ID"].astype(str).str.split('.',expand=True)[0]
schA["FORM_ID"]=schA["FORM_ID"].astype(str).str.split('.',expand=True)[0]

schA["KEY"] = schA["ACK_ID"].astype(str) + schA["FORM_ID"].astype(str).str.split('.',expand=True)[0]
schA['DROP']=np.where(schA["KEY"].isin(map["KEY"].tolist())==False,1,0)

#schA.loc[~schA["KEY"].isin(map["KEY"]), "DROP"] = 1
#schA.DROP.value_counts(dropna=False)
print("\nDrop b/c not n map ", schA[schA["DROP"] == 1].shape[0])
schA = schA[schA["DROP"] == 0].drop("DROP", axis=1)

schA["Carrier_Orig"] = map.set_index("KEY").loc[schA["KEY"], "Carrier_Orig"].values
schA["Carrier_EIN"] = map.set_index("KEY").loc[schA["KEY"], "Carrier_EIN"].values
schA["Carrier_NAIC"] = map.set_index("KEY").loc[schA["KEY"], "Carrier_NAIC"].values
schA["Members"] = map.set_index("KEY").loc[schA["KEY"], "Members"].values
schA["Premiums"] = map.set_index("KEY").loc[schA["KEY"], "Premiums"].values
schA["Medical"] = map.set_index("KEY").loc[schA["KEY"], "Medical"].values
schA["Dental"] = map.set_index("KEY").loc[schA["KEY"], "Dental"].values
schA["Vision"] = map.set_index("KEY").loc[schA["KEY"], "Vision"].values
schA["Life"] = map.set_index("KEY").loc[schA["KEY"], "Life"].values
schA["LTD"] = map.set_index("KEY").loc[schA["KEY"], "LTD"].values
schA["STD"] = map.set_index("KEY").loc[schA["KEY"], "STD"].values
schA["Stoploss"] = map.set_index("KEY").loc[schA["KEY"], "Stoploss"].values
schA["PBM"] = map.set_index("KEY").loc[schA["KEY"], "Drugs"].values
schA["CTR_NUM"] = map.set_index("KEY").loc[schA["KEY"], "CTR_NUM"].values

del map
schA = schA.drop(columns=["KEY"])

na_cols = schA.columns[schA.isna().any()].tolist()
if len(na_cols) > 0:
    for col in na_cols:
        dtype = str(schA[col].dtype)
        print(f"\nNA values of {col} type {dtype} {schA[col].isna().sum()}")
        if dtype == "object":
            schA.loc[schA[col].isna(), col] = ""
        elif dtype in ["int64", "float64"]:
            schA.loc[schA[col].isna(), col] = 0
missing_cols = list(set(schA.columns) - set(schC.columns))
for col in missing_cols:
    print(f"\nAdd {col} to schC set to")
    if schA[col].dtype == "object":
        print(" blank")
        schC[col] = ""
    else:
        print(" zero")
        schC[col] = 0
        

len(schC)
dt = pd.concat([schA, schC], ignore_index=True)
del schC
del schA


print(f"\nSaving brokers.all.yrs.Standardized.pkl at {brokerDir}")
dt.to_pickle(f"{brokerDir}brokers.all.yrs.Standardized.pkl")
print(f"\nObs saved in broker.all.yrs.Standardized. :{len(dt)}")

brokerNameManualAdjustments(
    inFile="brokers.all.yrs.Standardized", outFile="brokers.all.yrs.Standardized"
)

print("\nAll manual adjustments are done")
function for BadName check
def badNameCheck(dtCheck, var, txt=""):
checkBadMerge(dtCheck=dtCheck.copy(), txt=txt)


bads = ["%", "%H", "21Y", "HY"]
te = dtCheck[dtCheck[var].isin(bads)]
if te.shape[0] > 500:
    print(te)
    raise Exception("\nBroker variables got f-ed up " + txt)

varTerms = getVariousTerms(var=var)
print("\n\nObs w/ ", var, "=MISSING ", dtCheck[dtCheck[var] == "MISSING"].shape[0])
print("  obs w/ empty ", var, " ", dtCheck[dtCheck[var] == ""].shape[0])

print(
    " and obs w/ ",
    var,
    " in various Terms ",
    dtCheck[dtCheck[var].isin(varTerms)].shape[0],
)

print("\nUnique ", var, " values ", txt, dtCheck[var].nunique())
def brokerNameManualAdjustments(
inFile="brokers.all.yrs.Standardized", outFile="brokers.all.yrs.Standardized"
):
print("\nFor manual broker names adjustments, loading ", inFile, ".pkl")


dt = pd.read_pickle(os.path.join(datawd, "brokers", inFile + ".pkl"))
print("..obs..", dt.shape[0])
dt.to_pickle(os.path.join(datawd, "brokers", inFile + "_b4ManualAdj.pkl"))

print("\nLoading manual lookup table for broker")
broker_ManualLookup = pd.read_excel(
    os.path.join(xwalkDir, "brokers", "broker_ManualLookup.xlsx")
)

broker_ManualLookup["BrokerGiven"] = broker_ManualLookup["BrokerGiven"].str.upper()
broker_ManualLookup["BrokerProper"] = broker_ManualLookup[
    "BrokerProper"
].str.upper()

dt["BROKER"] = dt["BROKER"].str.upper()

oldObs = dt.shape[0]

# step 1, based on BROKER
dt2 = pd.merge(
    dt, broker_ManualLookup, left_on="BROKER", right_on="BrokerProper", how="left"
)

if len(dt2) != oldObs:
    raise Exception("\nRecords increased due to broker manual lookup")

dt2 = dt2.assign(
    BROKER=np.where(dt2.BrokerGiven.isna(), dt2.BROKER, dt2.BrokerGiven)
).copy()

dt2 = dt2.assign(
    BROKER=np.where(
        dt2.BROKER.isin(
            ["VHA MID AMERICA", "CHILD HEALTH CORP", "BCBS", "BCBS ND"]
        ),
        "",
        dt2.BROKER,
    )
).copy()

dt2 = dt2.drop(columns=["BrokerGiven"]).copy()

dt2 = dt2.assign(BrokerProper=dt2.BROKER.apply(lambda x: str.title(x.lower())))

# step 1, based on BROKER
# Summit fix for specific broker states
summitVariations = [
    "Summit",
    "Hilb Md Dba Summit",
    "Summit Insservices",
    "Summit Va",
    "Summit Insuarnace",
]
statesForChange = ["VA", "WV"]
toChange = "The Hilb Group"

dt2["BrokerProper"] = np.where(
    (dt2["BrokerProper"].isin(summitVariations))
    & (dt2["BROKER_STATE"].isin(statesForChange)),
    toChange,
    dt2["BrokerProper"],
)

dt2["BrokerProper"] = np.where(
    dt2["BrokerProper"].str.contains("Hrh "), "Wtw", dt2["BrokerProper"]
)

# Name specific spot fix
dt2["BrokerProper"] = np.where(
    (
        dt2["BROKER_INITIAL"].str.contains("^STRATEGIC EMPLOYEE BENEFIT SERVIC")
        | dt2["BROKER_INITIAL"].str.contains("STRATEGIC EMPLOYEE BENEFIT SVS")
        | dt2["BROKER_INITIAL"].str.contains("STRATEGIC EMPLO")
    )
    & (dt2["BrokerProper"] == "Hub"),
    "Strategic Employee Benefits Services",
    dt2["BrokerProper"],
)

dt2["BrokerProper"] = np.where(
    (
        dt2["BROKER_INITIAL"].str.contains("KEYSTONE BENEFITS")
        | dt2["BROKER_INITIAL"].str.contains("KEYSTONE INS")
    )
    & (~dt2["BROKER_INITIAL"].str.contains("ANDREW")),
    "Keystone Ins Benefits Group",
    dt2["BrokerProper"],
)

dt2["BrokerProper"] = np.where(
    (
        dt2["BROKER_INITIAL"].str.contains("STERLING GRO")
        | dt2["BROKER_INITIAL"].str.contains("STERLING INSURANCE GR")
        | dt2["BROKER_INITIAL"].str.contains("STERLING ROUP")
    )
    & (~dt2["BROKER_INITIAL"].str.contains("AMERICAN")),
    "Sterling Group",
    dt2["BrokerProper"],
)

# Cornerstone fix for specific broker states
# NOTE: McKinsey Changes
# dt2["BrokerProper"] = np.where(
#     (dt2["BROKER_STATE"] == "MO")
#     & (dt2["BROKER_INITIAL"].str.contains("CORNERSTONE"))
#     & (~dt2["BROKER_INITIAL"].str.contains("FIN")),
#     "AssuredPartners",
#     dt2["BrokerProper"],
# )

dt2["BrokerProper"] = np.where(
    (dt2["BROKER_STATE"] == "RI")
    & (dt2["BROKER_INITIAL"].str.contains("CORNERSTONE"))
    & (dt2["BROKER_ADDRESS"].str.contains("JEFFE")),
    "The Hilb Group",
    dt2["BrokerProper"],
)

dt2["BrokerProper"] = np.where(
    (dt2["BROKER_STATE"] == "OH")
    & (dt2["BROKER_INITIAL"].str.contains("CORNERSTONE"))
    & (dt2["BROKER_ADDRESS"].str.contains("FLORENCE")),
    "Cornerstone",
    dt2["BrokerProper"],
)

dt2["BrokerProper"] = np.where(
    (dt2["BROKER_STATE"] == "OH")
    & (dt2["BROKER_INITIAL"].str.contains("CORNERSTONE"))
    & (dt2["BROKER_INITIAL"].str.contains("BROK")),
    "Cornerstone",
    dt2["BrokerProper"],
)

dt2["BrokerProper"] = np.where(
    (dt2["BROKER_STATE"] == "")
    & (dt2["BROKER_INITIAL"].str.contains("CORNERSTONE"))
    & (dt2["BROKER_INITIAL"].str.contains("ASSURED")),
    "Assuredpartners Cornerstone, LLC",
    dt2["BrokerProper"],
)

dt2["BrokerProper"] = np.where(
    (dt2["BROKER_STATE"] == "PA")
    & (dt2["BROKER_INITIAL"].str.contains("CORNERSTONE"))
    & (dt2["BROKER_INITIAL"].str.contains("FIN"))
    & (dt2["BROKER_ADDRESS"].str.contains("HAMILTON")),
    "Cornerstone Financial",
    dt2["BrokerProper"],
)

dt2["BrokerProper"] = np.where(
    (dt2["BROKER_STATE"] == "PA")
    & (dt2["BROKER_INITIAL"].str.contains("CORNERSTONE"))
    & (dt2["BROKER_INITIAL"].str.contains("ADV")),
    "Cornerstone Adv Asset Management",
    dt2["BrokerProper"],
)

dt2["BrokerProper"] = np.where(
    (dt2["BROKER_STATE"] == "PA")
    & (dt2["BROKER_INITIAL"].str.contains("CORNERSTONE"))
    & (dt2["BROKER_INITIAL"].str.contains("INST")),
    "Cornerstone Adv Asset Management",
    dt2["BrokerProper"],
)

dt2["BrokerProper"] = np.where(
    (dt2["BROKER_STATE"] == "MD")
    & (dt2["BROKER_INITIAL"].str.contains("CORNERSTONE"))
    & (dt2["BROKER_INITIAL"].str.contains("ADV"))
    & (dt2["BROKER_ADDRESS"].str.contains("PADONIA")),
    "Cornerstone Advisory",
    dt2["BrokerProper"],
)

dt2["BrokerProper"] = np.where(
    (dt2["BROKER_STATE"] == "MD")
    & (dt2["BROKER_INITIAL"].str.contains("CORNERSTONE"))
    & (dt2["BROKER_INITIAL"].str.contains("FIN")),
    "Cornerstone Financial Maryland",
    dt2["BrokerProper"],
)
############ TODO Add new
dt2["BrokerProper"] = np.where(
    (dt2["BROKER_INITIAL"].str.contains("SALUS GROUP")),
    "Salus Group",
    dt2["BrokerProper"],
)
dt2["BrokerProper"] = np.where(
    (
        (dt2["BROKER_INITIAL"].str.contains("OLSON INSURANCE"))
        & (dt2["BROKER_INITIAL"].str.contains("NORMAN"))
    )
    | (dt2["BROKER_INITIAL"].str.contains("OLSON BENEFIT GROUP"))
    | (dt2["BROKER_INITIAL"].str.contains("OLSON INSURANCE GROUP"))
    | (
        (dt2["BROKER_INITIAL"].str.contains("OLSON INSURANCE "))
        & (dt2["BROKER_INITIAL"].str.contains("FINANCIAL"))
    )
    | (dt2["BROKER_INITIAL"] == ("THE OLSON GROUP")),
    "The Olson Group",
    dt2["BrokerProper"],
)
dt2["BrokerProper"] = np.where(
    (dt2["BROKER_INITIAL"].str.contains("BURNHAM"))
    & (dt2["BROKER_INITIAL"].str.contains("ALERA")),
    "Alera Group",
    dt2["BrokerProper"],
)
# End changes

# Matching the BROKER and BrokerProper columns
dt2["BROKER"] = dt2["BrokerProper"].str.upper()
# NOTE: McKinsey Changes here
dt2.loc[
    (dt2["BROKER_ZIPCODE"] != "")
    & (dt2["BROKER_ZIPCODE"] != "0")
    & (dt2["BROKER_ZIPCODE"].notna()),
    "BROKER_ZIPCODE",
] = dt2["BROKER_ZIPCODE"].str.zfill(5)
dt2.loc[
    (dt2["Zipcode"] != "") & (dt2["Zipcode"] != "0") & (dt2["Zipcode"].notna()),
    "Zipcode",
] = dt2["Zipcode"].str.zfill(5)
# End changes

if len(dt2.loc[dt2["BrokerProper"].str.upper() != dt2["BROKER"]]) > 0:
    raise Exception("\nNames in fields BROKER and BrokerProper do not match")

dt2.to_pickle(datawd + "brokers/" + outFile + ".pkl")

in the above pasted code if 
load_broker_standardization_parameters_with_states()  this is present can u please give me that code snippet and along with dry run can u please explain that to me