Iti0210w101demo

Peidetud Markovi mudeli näide loengust

Parameetrid (ülemineku- ja emissioonitõenäosused) ja abifunktsioon valimi võtmiseks.

import random

prior = {
   "cheat" : 0.5,
   "fair" : 0.5
}

transition = {
   "cheat" : {"fair" : 0.2, "cheat" : 0.8},
   "fair" : {"fair" : 0.7, "cheat" : 0.3},
}

emission = {
    "fair" : { "1" : 1/6, "2" : 1/6,"3" : 1/6,"4" : 1/6,"5" : 1/6,"6" : 1/6},
    "cheat" : { "1" : 0, "2" : 0,"3" : 0,"4" : 0,"5" : 0.2,"6" : 0.8}
}

def sample(d):
    population = []
    weights = []
    for k, v in d.items():
        population.append(k)
        weights.append(v)
    return random.choices(population, weights=weights)[0]

Genereerime mudeliga juhusliku seeria.

X = []
E = []
for i in range(50):
    if i == 0:
        Xi = sample(prior)
        print("picking", Xi, "die")
    else:
        prev = X[-1]
        Xi = sample(transition[prev])
        if prev != Xi:
            print("swapping die to", Xi)
    Ei = sample(emission[Xi])
    print(Ei)
    X.append(Xi)
    E.append(Ei)

Edasi-tagasi sõnumite arvutamine ja normaliseerimine:

# let's use a tuple for distributions
#  { "fair": 0.6, "cheat": 0.4} ==>  (0.6, 0.4)

def forward(fw_prev, e_k):
    # returns distribution over X_{k+1}
    result = []
    for x_k in ["fair", "cheat"]:
        S = 0
        for i, x_prev in enumerate(["fair", "cheat"]):
            S += transition[x_prev][x_k] * fw_prev[i]
        result.append(S * emission[x_k][e_k])
    res_S = sum(result)
    return tuple([r/res_S for r in result])

def backward(b_next, e_next):
    result = []
    for x_k in ["fair", "cheat"]:
        S = 0
        for i, x_next in enumerate(["fair", "cheat"]):
            S += emission[x_next][e_next] * transition[x_k][x_next] * b_next[i]
        result.append(S)
    return tuple(result)

def normalize(vect):
    S = sum(vect)
    return tuple([v/S for v in vect])

Peidetud muutujate leidmine.

def smoothed_estimate(E, t=50):
    f_prev = (prior["fair"], prior["cheat"])
    f_X = []
    for k in range(t):
        f_k = forward(f_prev, E[k])
        f_X.append(f_k)
        f_prev = f_k

    b = (0.5, 0.5)
    smooth_X = []
    for k in range(t-1, -1, -1):
        smooth_X.append(normalize([f_X[k][0] * b[0], f_X[k][1] * b[1]]))
        b = backward(b, E[k])
    smooth_X.reverse()
    return smooth_X

s_X = smoothed_estimate(E)
for i in range(50):
    print(X[i], E[i], "cheat probability", s_X[i][1])

Tulemuse hindamine.

def evaluate(guess_X, real_X, threshold=0.7):
    TP = 0
    FN = 0
    FP = 0
    for i in range(len(guess_X)):
        if guess_X[i][1] > threshold:
            if real_X[i] == "cheat":
                TP += 1
            else:
                FP += 1
        else:
            if real_X[i] == "cheat":
                FN += 1

    print("")
    print("Cheating caught {:.0f}%".format((TP/(TP+FN))*100))
    print("False accusations {:.0f}%".format((FP/(TP+FP))*100))

evaluate(s_X, X)

EM-algoritmiga üleminekujaotuse leidmine.

transition_orig = transition
transition = {}
for x in ["fair", "cheat"]:
    p = random.random()
    transition[x] = {"fair": p, "cheat": 1-p}

from collections import defaultdict

def em(E, iters=100):
    for i in range(iters):
        expected_X = smoothed_estimate(E)
        virtual_t = {"fair" : defaultdict(float),
                  "cheat" : defaultdict(float)}
        virtual_N = {"fair" : 0.0, "cheat" : 0.0}
        for j in range(len(E)-1):
            virtual_N["fair"] += expected_X[j][0]
            virtual_t["fair"]["fair"] += expected_X[j][0] * expected_X[j+1][0]
            virtual_t["cheat"]["fair"] += expected_X[j][1] * expected_X[j+1][0]
        virtual_N["cheat"] = len(E) - 1 - virtual_N["fair"]
        for x in ["fair", "cheat"]:
            p = virtual_t[x]["fair"]/virtual_N[x]
            transition[x] = {"fair": p, "cheat": 1-p}
        print(transition)
    return expected_X

em_X = em(E, 20)
evaluate(em_X, X)