pv-current/test_one_unit.py

# Created by: leo
# Created on: 2018.10.26

from unittest import TestCase, defaultTestLoader, TextTestRunner
from datetime import datetime, timedelta
from sklearn.cluster import KMeans
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from max_current import (coef_maxcur, string_coef_maxcur, max_current_monthly,
                         calc_maxcur_coefs, train_thresholds, classifier)
from coef_matrix import (train_median_thresholds, string_coef_matrix,
                         calc_matrix_coefs, coef_median)
import logging

logging.basicConfig(level=logging.DEBUG)


class ForecastTest(TestCase):
    def setUp(self):
        self.raw = pd.read_csv('ycz6502.csv', usecols=[0, 4, 6],
                               index_col=0).dropna()
        self.raw.index = pd.to_datetime(self.raw.index)
        self.daily = self.raw.groupby(pd.Grouper(freq='D'))

    def test_zero_timeseries(self):
        pass


class AnomalyDetectionTest(TestCase):
    def setUp(self):
        self.raw = pd.read_csv('ycz6502.csv', usecols=[0, 4, 6],
                               index_col=0).dropna()
        self.raw.index = pd.to_datetime(self.raw.index)
        self.daily = self.raw.groupby(pd.Grouper(freq='D'))

    def test_coef_of_day(self):
        daydata = self.daily.get_group('2017-04-08')
        logging.info(coef_maxcur(daydata))
        # logging.info(coef_median(daydata))

    def test_coef_of_days(self):
        wins = ['2017-05-28', '2017-05-29', '2017-06-03', '2017-06-04']
        for win, coefs in zip(wins, map(lambda x: coef_maxcur(self.daily.get_group(x)), wins)):
            logging.info('%s\n%s' % (win, coefs))

    def test_str_coef_some_days(self):
        wins = ['2017-05-24', '2017-05-25', '2017-05-26', '2017-05-27',
                '2017-05-28', '2017-05-29', '2017-05-30', '2017-05-31',
                '2017-06-01', '2017-06-02', '2017-06-03', '2017-06-04',
                '2017-06-06', '2017-06-07']
        cluster_id = 1
        cluster_status = {}
        for item in zip(wins, map(lambda x: coef_maxcur(self.daily.get_group(x)), wins)):
            cluster_status[item[0]] = item[1][cluster_id]
        lists = sorted(cluster_status.items())
        x, y = zip(*lists)
        plt.scatter(x, y)
        plt.xticks(rotation=70)
        plt.title('Coefficients of String in Specific Days')
        plt.show()

    def test_maxcur_coef_all_days(self):
        string_id = 1
        coefs = self.daily.apply(string_coef_maxcur, string_id)
        plt.scatter(coefs.index, coefs, c='g', alpha=0.5)
        plt.xlabel('Date')
        plt.ylabel('String Coefficients')
        plt.title('Coefficients of String %d in All Days(Max Current)' % string_id)
        plt.grid()
        plt.show()

    def test_matrix_coef_all_days(self):
        string_id = 1
        coefs = self.daily.apply(string_coef_matrix, string_id)
        plt.scatter(coefs.index, coefs, c='g', alpha=0.5)
        plt.xlabel('Date')
        plt.ylabel('String Coefficients')
        plt.title('Coefficients of String %d in All Days (Matrix Median)' % string_id)
        plt.grid()
        plt.show()

    def test_daily_plot(self):
        cur528 = self.raw['2017-05-28']
        cur528_c1 = cur528[cur528['zuchuan'] == 1]['value']
        start = datetime(2017, 5, 28, 6, 0)
        end = datetime(2017, 5, 28, 19, 45)
        step = timedelta(minutes = 15)
        plt.plot(cur528_c1.index, cur528_c1)
        plt.title('Coeffients of String in a Time Span')
        plt.show()

    def test_groups(self):
        self.daily.aggregate(coef_maxcur)
        self.daily.aggregate(np.sum)
        for name, grp in self.daily:
            logging.info('%s\n%s' % (name, coef_maxcur(grp)))


class TrainerTest(TestCase):
    def setUp(self):
        self.raw = pd.read_csv('ycz6502.csv', usecols=[0, 4, 6],
                               index_col=0).dropna()

    def test_monthly_max(self):
        res = max_current_monthly(self.raw)
        self.assertAlmostEqual(res['2017-03'][0], 9.75)

    def test_trainer_03(self):
        logging.info('Thresholds between statuses:\n%s' % train_thresholds(self.raw, 0.3))

    def test_trainer_08(self):
        logging.info('Thresholds between statuses:\n%s' % train_thresholds(self.raw, 0.8))

    def test_trainer_095(self):
        logging.info('Thresholds between statuses:\n%s' % train_thresholds(self.raw, 0.95))

    def test_matrix_trainer(self):
        logging.info('Thresholds between statuses:\n%s' % train_median_thresholds(self.raw))

    def test_maxcur_all_coefs_plot(self):
        coefs = calc_maxcur_coefs(self.raw, 0.8)
        for col in coefs.columns:
            plt.scatter(coefs.index, coefs[col], alpha=0.95, s=10, marker='x')
        plt.xlabel('Date')
        plt.ylabel('Coefficients')
        plt.title('Coefficients of all strings with max-current method')
        plt.grid()
        plt.show()

    def test_matrix_all_coefs_plot(self):
        coefs = calc_matrix_coefs(self.raw)
        for col in coefs.columns:
            plt.scatter(coefs.index, coefs[col], alpha=0.95, s=10, marker='x')
        plt.xlabel('Date')
        plt.ylabel('Coefficients')
        plt.title('Coefficients of all strings with matrix-median method')
        plt.grid()
        plt.show()


class ClassifierTest(TestCase):
    def setUp(self):
        self.raw = pd.read_csv('ycz6502.csv', usecols=[0, 4, 6],
                               index_col=0).dropna()
        self.raw.index = pd.to_datetime(self.raw.index)
        self.daily = self.raw.groupby(pd.Grouper(freq='D'))

    def test_low_current(self):
        data0323 = self.daily.get_group('2017-03-24')
        logging.info(classifier(data0323, 8))

    def test_normal_current(self):
        data0605 = self.daily.get_group('2017-06-05')
        logging.info(classifier(data0605, 4))

    def test_warn_status(self):
        data0528 = self.daily.get_group('2017-05-28')
        logging.info(classifier(data0528, 8))

    def test_watch_status(self):
        data0604 = self.daily.get_group('2017-06-04')
        logging.info(classifier(data0604, 8))


class UtilTest(TestCase):
    def test_combine_sets(self):
        inp = {1: {3, 2, 5}, 2: {5, 8, 7}, 3: {23, 34, 2}}
        res = set().union(*inp.values())
        self.assertEqual(res, {2, 7, 3, 34, 5, 23, 8})

    def test_kmeans(self):
        inp = {1, 0.9, 0.92, 0.89, 0.4, 0.2, 0.6, -0.2, 0.84, -0.05}
        data = np.array(list(inp)).reshape(-1, 1)
        kmeans = KMeans(n_clusters=3, random_state=1)   # the random_state make the group ID in clustering result fixed
        kmeans.fit(data)
        res = kmeans.predict(data)
        np.testing.assert_array_equal(res, [1, 1, 1, 1, 2, 2, 2, 0, 1, 0])


if __name__ == '__main__':
    # tests = defaultTestLoader.loadTestsFromTestCase(AnomalyDetectionTest)
    # logging.info(tests.countTestCases())
    test_trainer = defaultTestLoader.loadTestsFromName(
            'test_app.TrainerTest.test_trainer_08')
    TextTestRunner(verbosity=1).run(test_trainer)