notebook.community

Edit and run

KNN & DTW





In [ ]:



    


# -*- coding: utf-8 -*-
class Dtw(object):
    
    def __init__(self, seq1, seq2,
                 patterns = [(-1,-1), (-1,0), (0,-1)], 
                 weights = [{(0,0):2}, {(0,0):1}, {(0,0):1}], 
                 band_r=0.005):
        self._seq1 = seq1
        self._seq2 = seq2
        self.len_seq1 = len(seq1)
        self.len_seq2 = len(seq2)
        self.len_pattern = len(patterns)
        self.sum_w = [sum(ws.values()) for ws in weights]
        self._r = int(len(seq1)*band_r)
        assert len(patterns) == len(weights)
        self._patterns = patterns
        self._weights = weights
    
    def get_distance(self, i1, i2):
        return abs(self._seq1[i1] - self._seq2[i2])

    def calculate(self):
        g = list([float('inf')]*self.len_seq2 for i in range(self.len_seq1))
        cost = list([0]*self.len_seq2 for i in range(self.len_seq1))

        g[0][0] = 2*self.get_distance(0, 0)
        for i in range(self.len_seq1):
            for j in range(max(0,i-self._r), min(i+self._r+1, self.len_seq2)):
                for pat_i in range(self.len_pattern):
                    coor = (i+self._patterns[pat_i][0], j+self._patterns[pat_i][1])
                    if coor[0]<0 or coor[1]<0:
                        continue
                    dist = 0
                    for w_coor_offset, d_w in self._weights[pat_i].items():
                        w_coor = (i+w_coor_offset[0], j+w_coor_offset[1])
                        dist += d_w*self.get_distance(w_coor[0], w_coor[1])
                    this_val = g[coor[0]][coor[1]] + dist
                    this_cost = cost[coor[0]][coor[1]] + self.sum_w[pat_i]
                    if this_val < g[i][j]:
                        g[i][j] = this_val
                        cost[i][j] = this_cost
        return g[self.len_seq1-1][self.len_seq2-1]/cost[self.len_seq1-1][self.len_seq2-1], g, cost
    
    def print_table(self, tb):
        print('      '+' '.join(["{:^7d}".format(i) for i in range(self.len_seq2)]))
        for i in range(self.len_seq1):
            str = "{:^4d}: ".format(i)
            for j in range(self.len_seq2):
                str += "{:^7.3f} ".format(tb[i][j])
            print (str)

    def print_g_matrix(self):
        _, tb, _ = self.calculate()
        self.print_table(tb)

    def print_cost_matrix(self):
        _, _, tb = self.calculate()
        self.print_table(tb)
        
    def get_dtw(self):
        ans, _, _ = self.calculate()
        return ans



    











In [ ]:



    


import csv
import random
import math
import operator
import numpy as np

def loadDataset(filename, data=[]):
    with open(filename, 'rb') as csvfile:
        lines = csv.reader(csvfile,delimiter=' ')
        dataset = list(lines)
        for x in range(len(dataset)):
            dataset[x] = filter(None, dataset[x])
            dataset[x] = list(map(float, dataset[x]))
            data.append(dataset[x])

def euclideanDistance(instance1, instance2, length):
	distance = 0
	for x in range(length):
		if x == 0:
			continue
		distance += pow((instance1[x] - instance2[x]), 2)
	return math.sqrt(distance)
 
def getNeighbors(trainingSet, testInstance, k, pattern, weight, band_r):
	distances = []
	length = len(testInstance)
	for x in range(len(trainingSet)):
#  z-normalization
		new_testInstance = (np.array(testInstance)-np.mean(testInstance))/np.std(testInstance)
		new_trainingSet = (np.array(trainingSet[x])-np.mean(trainingSet[x]))/np.std(trainingSet[x])
		d = Dtw(new_testInstance[1:], new_trainingSet[1:], pattern, weight, band_r)
		dist = d.get_dtw()
# 		dist = euclideanDistance(testInstance, trainingSet[x], length)
		distances.append((trainingSet[x], dist))
	distances.sort(key=operator.itemgetter(1))
#  	print "dist >>>> ",distances
	neighbors = []
	for x in range(k):
		neighbors.append(distances[x][0])
	return neighbors

def getResponse(neighbors):
	classVotes = {}
	for x in range(len(neighbors)):
		response = neighbors[x][0]
		if response in classVotes:
			classVotes[response] += 1
		else:
			classVotes[response] = 1
	sortedVotes = sorted(classVotes.iteritems(), key=operator.itemgetter(1), reverse=True)
	return sortedVotes[0][0]
 
def getAccuracy(testSet, predictions):
	correct = 0
	for x in range(len(testSet)):
		if testSet[x][0] == predictions[x]:
			correct += 1
	return (correct/float(len(testSet))) * 100.0
	
def knn(train_data, test_data, k, pattern, weight, band_r):
	# prepare data
	trainingSet=[]
	testSet=[]
	loadDataset(train_data, trainingSet)
	loadDataset(test_data, testSet)
# 	print 'Train set: ' + repr(len(trainingSet))
# 	print trainingSet
# 	print 'Test set: ' + repr(len(testSet))
# 	print testSet
	# generate predictions
	predictions=[]
	for x in range(len(testSet)):
# 		print ">>",testSet[x]
		neighbors = getNeighbors(trainingSet, testSet[x], k, pattern, weight, band_r)
# 		print "neighbors >>", neighbors
		result = getResponse(neighbors)
# 		print "result >>", result
		predictions.append(result)
# 		print('> predicted=' + repr(result) + ', actual=' + repr(testSet[x][0]))
	accuracy = getAccuracy(testSet, predictions)
	return accuracy

Main





In [ ]:



    


PATTERNS_1 = [(0,-1), (-1,-1), (-1,0)]
WEIGHTS_SYM_1 = [{(0,0):1}, {(0,0):2}, {(0,0):1}] 
import numpy as np



    











In [ ]:



    


TRAIN_DATA = 'dataset/Coffee_TRAIN'
TEST_DATA = 'dataset/Coffee_TEST'
accuracy = []
accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.005) )
# accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.015) )
# accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.025) )
print accuracy
print np.mean(accuracy)



    











In [ ]:



    


TRAIN_DATA = 'dataset/Beef_TRAIN'
TEST_DATA = 'dataset/Beef_TEST'
accuracy = []
accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.005) )
# accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.015) )
# accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.025) )
print accuracy
print np.mean(accuracy)



    











In [ ]:



    


TRAIN_DATA = 'dataset/FISH_TRAIN'
TEST_DATA = 'dataset/FISH_TEST'
accuracy = []
accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.005) )
# accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.015) )
# accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.025) )
print accuracy
print np.mean(accuracy)



    











In [ ]:



    


TRAIN_DATA = 'dataset/wafer_TRAIN'
TEST_DATA = 'dataset/wafer_TEST'
accuracy = []
accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.005) )
# accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.015) )
# accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.025) )
print accuracy
print np.mean(accuracy)



    











In [ ]:



    


TRAIN_DATA = 'dataset/FaceAll_TRAIN'
TEST_DATA = 'dataset/FaceAll_TEST'
accuracy = []
accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.005) )
# accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.015) )
# accuracy.append( knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, WEIGHTS_SYM_1, 0.025) )
print accuracy
print np.mean(accuracy)



    











In [ ]:



    


with open(OUTPUT_FILE, "w") as myfile:
    myfile.write("i,j,k,accuracy\n")
for weight in weights:
    i = weight[0][(0,0)]
    j = weight[1][(0,0)]
    k = weight[2][(0,0)]
    print "i:", i, "j:", j,"k:", k
    acc = knn(TRAIN_DATA, TEST_DATA, 1, PATTERNS_1, weight)
    print acc
    with open(OUTPUT_FILE, "a") as myfile:
        myfile.write(str(i)+","+str(j)+","+str(k)+","+str(acc)+"\n")



    











In [ ]:

Content source: Roos12005/time-series-dtw

Similar notebooks:

notebook.community | gallery | about