Arbeidsfradrag: Hvordan gjøres trekningen?

"""
Dette programmet trekker tilfeldig 8 prosent av personene i målgruppen som skal få rett på arbeidsfradrag (gruppe A). Trekningen gjennomføres på en måte som sikrer jevn fordeling etter skattepliktstatus, bosattstatus, positiv lønnsinntekt i 2024, positiv trygdeinntekt i 2025, positiv næringsinntekt i 2024, kjønn, fødselsår og lønnsinntekt i 2025. Blant de som ikke trekkes til å få rett på fradrag, trekkes det en gruppe som får ekstra informasjon om skattereglene (gruppe B). Resten av målgruppen får verken fradrag eller ekstra informasjon (gruppe C).

Trekningen er fullt reproduserbar fordi den benytter en fast "nøkkel" (startverdi for tilfeldig trekning). 

Denne nøkkelen lages fra personenes ID-er i data kombinert med en ekstern nøkkel kalt "nonce". 

For å sikre at trekningen ikke kan manipuleres skal det gjøres følgende: 

(1) Innen en angitt dato (31.okt, 2025) kjøres funksjonen digest(data) på datasettet. Denne gir en kontrollnøkkel bestående av 8 tall som skrives ned, deles eksternt, og føres inn i koden på "DATASETHASH = np.array([..]"). Disse tallene er kun de samme dersom datasettet er det samme. Når trekningen skal gjennomføres må datasettet være uendret for at koden skal kjøre, og kodenøklene skal sjekkes opp mot de nedskrevne og delte kodenøklene fra 31.okt. 
(2) I trekningen benyttes en kodenøkkel 2 ("nonce") som per 31.okt ikke er realisert (ikke finnes). Her benyttes en børskurs for en senere dato. 

Datasettet er dermed låst før den andre kodenøkkelen (som avgjør utfallet) finnes. 
 """

# Importerer biblioteker som er nødvendige for å kjøre koden
import pandas as pd
import numpy as np
import hashlib
import pickle
import itertools

# hash av datasettet. Det er 256 bits, organisert som 8 uint32. Har man et
# datasett `df` bruker man funksjonen `digest` definert nedenfor: `digest(df)`
# for å finne hash'en.  Riktig hash settes så inn i variablen DATASETHASH nedenfor
# Den testes i `stratifiedsample`.

DATASETHASH = np.array(["her skal kontrollnøkkelen fra datasettet legges inn"], dtype='uint32')

def digest(data):
    """
        Lager en 256 bit hash fra data. Returnerer en 8 lang numpy array av uint32.
    """
    shahash = hashlib.sha3_256(pickle.dumps(data, protocol=4)).digest()
    return np.frombuffer(shahash, dtype='uint32')

def stratifiedsample(df, *, size=100, numdraw=8, numdraw2="Antall i hver gruppe som får ekstra informasjon",
                     nonce="Åpningskursen på OSEBX 3. november"):
    """Trekker et stratifisert tilfeldig utvalg. Stratifiseringen er
    over variablene 'faar' og 'kjonn' og kategorier beregnet fra kovariatene.
    Innenfor disse gruppene grupperes det grupper på 100, og det trekkes åtte fra hver.

    Utvalget er tilfeldig, men deterministisk på den måten at seed'en
    til random-generatoren beregnes fra `df['id']` og `nonce`.

    Argumenter
    ----------
    df : en :class:`pandas.DataFrame` med kolonner 'id', 'faar', 'kjonn', 'bosatt',
        'skatteplikt', 'W2025', 'W2024', 'B2025', 'N2024'
        Kolonnen 'id' må være unik.

    size : størrelsen på inntektsgrupper i hvert stratum. Default: 100

    numdraw : Antall å trekke til fradrag fra hver inntektsgruppe. Til gruppe 'A'. Default 8.

    numdraw2 : Antall å trekke til ekstra informasjon fra hver inntektsgruppe. Til gruppe 'B'. 

    nonce : et vilkårlig python objekt som sammen med id brukes for å
        lage en seed til random-generatoren som gjør utvalget.

    Returverdi
    ----------
    :class:`pandas.DataFrame`
        En kopi av df, med kolonner lagt til:
          'kategori' : 7-delt kategori basert på kovariatene.
          'inntektsgruppe' : inntektsgruppe innenfor faar, kjønn, kategori
          'G' : id for gruppene det er trukket fra.
          'trekning' : Indikator for om denne personen er trukket ut til A, B eller C.

        Sortert på indeksen fra det originale datasettet.

    """

    # sjekk at datasettet har samme hash som før
    hash = digest(df)
    if not all(hash == DATASETHASH):
        raise RuntimeError("Datasettets hash stemmer ikke")

    #Sjekk at datasettet bare har unike personlige løpenumre og sorter etter dette
    if not df.id.is_unique:
        raise RuntimeError("column 'id' is not unique")

    fraction = numdraw/size

    df = df.sort_values(by='id')

    # Deler målgruppen i 7 kategorier basert på skattepliktstatus, bosattstatus, positiv lønnsinntekt i 2024, positiv trygdeinntekt i 2025, positiv næringsinntekt i 2024.
    df['kategori'] = pd.Categorical([0 for _ in range(len(df))], ('K1','K2','K3','K4','K5','K6','K7'))
    df.loc[(df.skatteplikt==1) & (df.bosatt==1) & (df.W2024==1) & (df.B2025==0) & (df.N2024==0), 'kategori'] = 'K1'
    df.loc[(df.skatteplikt==1) & (df.bosatt==1) & (df.W2024==1) & (df.B2025==1) & (df.N2024==0), 'kategori'] = 'K2'
    df.loc[(df.skatteplikt==1) & (df.bosatt==1) & (df.W2024==0) & (df.B2025==0) & (df.N2024==0), 'kategori'] = 'K3'
    df.loc[(df.skatteplikt==1) & (df.bosatt==1) & (df.W2024==0) & (df.B2025==1) & (df.N2024==0), 'kategori'] = 'K4'
    df.loc[(df.skatteplikt==1) & (df.bosatt==1) & (df.N2024==1), 'kategori'] = 'K5'
    df.loc[(df.skatteplikt==1) & (df.bosatt==0), 'kategori'] = 'K6'
    df.loc[(df.skatteplikt==0), 'kategori'] = 'K7'

    # gruppér i undergrupper (strata) på 100 personer basert på fødselsår, kjønn, kategori og stigende lønnsinntekt i 2025
    # Lag grupper (strata) etter fødselsår, kjønn og kategori.
    # Innen hver undergruppe sorteres personene etter lønnsinntekt hittil i 2025 og deles inn i grupper på 100 personer.

    g = df.groupby(['faar', 'kjonn', 'kategori'], observed=True, sort=False)

    # gruppe 0 er restgruppen hvis antallet ikke er multiplum av size (100)
    groups = g.apply(lambda grp: grp.groupby(_cutn(grp.W2025, size),
                                             observed=True, sort=False))

    df = df.assign(inntektsgruppe=pd.concat(list(groups.map(lambda grp: grp.keys))))

    df['G'] = pd.Categorical(map(lambda x: 'G' + '.'.join(map(str, x)),
                                         zip(df.inntektsgruppe, df.faar, df.kjonn, df.kategori)))

    ### TILFELDIG TREKNING
    # Oppretter en tallgenerator som gir samme resultat hver gang basert på alle id-numre og nonce.
    rng = np.random.Generator(np.random.MT19937(digest((np.array(df.id), nonce))))

    # trekk til gruppe 'A' og 'B'
    idxA = df[df.inntektsgruppe != 0].groupby(by='G', observed=True, sort=False).sample(n=numdraw, random_state=rng).index
    dfnotA = df.loc[~df.index.isin(idxA)]
    idxB = dfnotA[dfnotA.inntektsgruppe != 0].groupby(by='G', observed=True, sort=False).sample(n=numdraw2, random_state=rng).index
    # trekk andel fra gruppe 0, altså resten hvis antallet ikke gikk opp i 100.
    if sum(df.inntektsgruppe==0) > 0:
        draw = rng.random(len(df))
        # trekk en andel numdraw/size fra restgruppen til A
        idxA = idxA.append(df[(df.inntektsgruppe==0) & (draw < numdraw/size)].index)
        # trekk en andel numdraw2/size fra de som er igjen til B
        dfnotA = df.loc[~df.index.isin(idxA)]
        draw = rng.random(len(dfnotA))
        idxB = idxB.append(dfnotA[(dfnotA.inntektsgruppe==0) & (draw < numdraw2/(size-numdraw))].index)

    # merk av status (A, B eller C) i en ny variabel 'trekning'
    df['trekning'] = pd.Categorical(['C' for _ in range(len(df))], ('A','B','C'))
    df.loc[idxA, 'trekning'] = 'A'
    df.loc[idxB, 'trekning'] = 'B'

    # add size of each group(N)
    oldindex = df.index
    N = df.groupby('G', observed=True).size()
    df.index = df.G
    df['N'] = N
    df.index = oldindex
    
# number individuals randomly - dersom det besluttes at forsøket skaleres ned til x pst. i gruppe A, brukes "randid" for å velge ut hvem som skal forbli i gruppe A. De x/8 i hver gruppe (strata) med lavest verdi på "randid" forblir i gruppe A. Resten av gruppen flyttes til gruppe C.

    seq = list(range(len(df)))
    rng.shuffle(seq)
    df['randid'] = seq

    df.sort_index(inplace=True)
    return df

# Funksjon som deler verdier i grupper på n (100) personer.
# Sorterer først, gir deretter gruppenummer 1, 2, 3, ...
# Resten som ikke fyller en hel gruppe får nummer 0.

def _cutn(g, size):
    """
    100 personer i hver gruppe 1 til N. Gjenværende personer i gruppe 0.
    """
    g = g.sort_values()
    numberofbins = len(g) // size
    binned = size*numberofbins
    remaining = len(g) - binned
    bins = pd.Series(sum( (list(itertools.repeat(i+1, size)) for i in range(numberofbins)), start=[]),
                     index=g.index[:binned])

    if remaining == 0:
        return bins

    rembins = pd.Series((0 for _ in range(remaining)), index=g.index[binned:])
    if len(bins) == 0:
        return rembins

    return pd.concat((bins, rembins))

---------- NY CELLE ----------

%python
#Her kjøres selve trekningen. nonce settes til åpningskurs på Oslo børs 3. november 2025
trekning = stratifiedsample(maalgruppe, nonce=42)

display(trekning)

---------- NY CELLE ----------

%python
if isinstance(trekning, pd.DataFrame):
    spark_df = spark.createDataFrame(trekning)
else:
    spark_df = trekning  # antar det er Spark DF

spark_df.write.mode("overwrite").saveAsTable("trekningsresultat")