EI328 Final Project Review (III)

　　At the end of this project, we are required to implement a basic classifier on our own and make use of it in the Min-Max Modular Neural Network. Here we made a Logistic Regression Classifier with naive batch gradient descent optimization method.

　　The discriminant function is a sigmoid function as follow:

　　　　　　$y(vec x)={1+exp{-(vec w^Tcdotvec x+b) }}^{-1}$

　　The objective function is the negative logarithm of likelihood:

　　　　　　$-ln L=sum_{n=1}^l ln{1+exp{-t_ncdot(vec w^Tcdotvec x+b)}}$

　　The gradient can be calculated as follow:

　　　　　　$ abla_{vec w}(-ln L)=sum_{n=1}^l{y(vec x_n)-frac{t_n+1}{2}}cdot vec x_n$

　　Here is the source code of ./src/DIY.java:

/**
* Argument Format:
* <task # = 6-2 or 6-3> <number of threads> <min group size> <max group size> <group size step>
**/

import java.util.concurrent.atomic.*;
import java.util.concurrent.*;
import java.util.*;
import java.io.*;

class Node {
    public int idx;
    public double val;
    
    public Node(int idx,double val) {
        this.idx = idx;
        this.val = val;
    }
}

class DataSet {
    /** This class mimic Problem class in LIBLINEAR **/
    public int l, n;
    public List<List<Node>> x;
    public List<Integer> y;
    
    public DataSet(int l,int n) {
        x = new ArrayList<List<Node>>();
        y = new ArrayList<Integer>();
        this.l = l;
        this.n = n;
    }
    public DataSet(String path) {
        x = new ArrayList<List<Node>>();
        y = new ArrayList<Integer>();
        StringTokenizer tok = null;
        StringTokenizer tok1 = null;
        Scanner in = null;
        try {
            in = new Scanner(new File(path));
            for (;in.hasNextLine();l++) {
                List<Node> tmp = new ArrayList<Node>();
                tok = new StringTokenizer(in.nextLine());
                y.add(Integer.valueOf(tok.nextToken()));
                while (tok.hasMoreTokens()) {
                    tok1 = new StringTokenizer(tok.nextToken(),":");
                    int idx = Integer.parseInt(tok1.nextToken());
                    double val = Double.parseDouble(tok1.nextToken());
                    tmp.add(new Node(idx,val));
                    if (idx>n) {
                        n = idx;
                    }
                }
                x.add(tmp);
            }
            in.close();
        } catch (Exception e) {
            System.err.println("DATASET Error: "+e);
        }
    }
    public int getY(int idx) {
        if (idx<0||idx>=l) {
            throw new RuntimeException("DataSet: IndexOutOfBound");
        }
        return y.get(idx).intValue();
    }
}

class Augur {
    /** This class mimic Model class in LIBLINEAR **/
    private static int id = 0;
    private final double lamb = 0;
    private DataSet prob;
    private double[] w;
    private double b;
    private int n;
    
    public Augur() {}
    public Augur(DataSet prob) {
        this.prob = prob;
        n = prob.n;
        w = new double[n];
        double[] g = new double[n+1];
        for (int itr=0;itr<1000;itr++) {
            calGrad(g);
            double len = norm(g);
            if (itr%20==0) {
                System.out.println("itr "+itr+": norm(g) = "+len);
            }
            //double step = optStep(g,len);
            double step = len;
            for (int i=0;i<n;i++) {
                w[i] -= step*g[i];
            }
            b -= step*g[n];
        }
        System.out.println("	Group "+(++id)+" Done");
    }
    public int predict(List<Node> x) {
        return (sigmoid(x)>0.5)? 1:-1;
    }
    private double optStep(double[] g,double len) {
        double[] val = new double[2];
        double p = 0, r = len;
        double q1 = .382*r-.618*p;
        double q2 = .618*r+.382*p;
        val[0] = error(q1,g);
        val[1] = error(q2,g);
        for (int i=0;i<5;i++) {
            if (val[0]<val[1]) {
                r = q2;
                q2 = q1;
                q1 = .382*r-.618*p;
                val[1] = val[0];
                val[0] = error(q1,g);
            } else {
                p = q1;
                q1 = q2;
                q2 = .618*r+.382*p;
                val[0] = val[1];
                val[1] = error(q2,g);
            }
        }
        return (val[0]<val[1])? q1:q2;
    }
    private double sigmoid(List<Node> x) {
        double val = b;
        for (Node node: x) {
            if (node.idx<=n) {
                val += w[node.idx-1]*node.val;
            }
        }
        return 1./(1+Math.exp(-val));
    }
    private double error(double step,double[] g) {
        double val = .5*lamb*square(w,b);
        for (int i=0;i<prob.l;i++) {
            double tmp = b-step*g[n];
            for (Node node:prob.x.get(i)) {
                tmp += (w[node.idx-1]-step*g[node.idx-1])*node.val;
            }
            val += Math.log(1+Math.exp(-prob.getY(i)*tmp));
        }
        return val;
    }
    private void calGrad(double[] g) {
        for (int i=0;i<n;i++) {
            g[i] *= w[i]*lamb;
        }
        g[n] = b*lamb;
        for (int i=0;i<prob.l;i++) {
            double delt = sigmoid(prob.x.get(i))-(prob.getY(i)+1)/2;
            for (Node node:prob.x.get(i)) {
                g[node.idx-1] += node.val*delt;
            }
            g[prob.n] += delt;
        }
    }
    private double square(double[] vect,double cst) {
        double val = cst*cst;
        for (int i=0;i<vect.length;i++) {
            val += vect[i]*vect[i];
        }
        return Math.sqrt(val);
    }
    private double norm(double[] vect) {
        return Math.sqrt(square(vect,0));
    }
}

class AbstractDIY extends Thread {
    protected static DataSet train, test;
    protected static PrintWriter out;
    protected static long start, end;
    
    protected static void init() {
        try {
            System.out.println();
            if (!(new File("./data/new_train.txt")).exists()) {
                System.out.println("Preprocessing: please wait a minute!");
                preproc("./data/","train.txt","");
                preproc("./data/","test.txt","5001:1.00");
                System.out.println("Ready!");
            }
            System.out.println("Reading the files ...	Wait please ~ ~");
            train = new DataSet("./data/new_train.txt");
            test = new DataSet("./data/new_test.txt");
        } catch (Exception e) {
            System.out.println("INIT Error: "+e);
        }
    }
    protected static void print(String str) {
        try {
            System.out.print(str);
            out.print(str);
        } catch (Exception e) {
            System.err.println("PRINT Error: "+e);
        }
    }
    protected static void println(String str) {
        try {
            System.out.println(str);
            out.println(str);
        } catch (Exception e) {
            System.err.println("PRINTLN Error: "+e);
        }
    }
    protected static void printTime(long time,boolean train) {
        try {
            if (train) {
                println("	Training:	"+time+"ms elapsed");
            } else {
                println("	Testing:	"+time+"ms elapsed");
            }
        } catch (Exception e) {
            System.err.println("PRINTTIME Error: "+e);
        }
    }
    protected static void stats(int[] res) {
        try {
            int truePos=0, falsePos=0, falseNeg=0, trueNeg=0;
            for (int i=0;i<test.l;i++) {
                if (test.getY(i)>0) {
                    if (res[i]>0) {
                        truePos++;
                    } else {
                        falseNeg++;
                    }
                } else {
                    if (res[i]>0) {
                        falsePos++;
                    } else {
                        trueNeg++;
                    }
                }
            }
            System.out.println("	"+truePos+"	"+falsePos+"	"+falseNeg+"	"+trueNeg);
            double acc = (truePos+trueNeg+.0)/test.l;
            println("	acc	= "+acc);
            double p = (truePos+.0)/(truePos+falsePos);
            double r = (truePos+.0)/(truePos+falseNeg);
            double f1 = 2*r*p/(r+p);
            println("	F1	= "+f1);
            double tpr = (truePos+.0)/(truePos+falseNeg);
            double fpr = (falsePos+.0)/(falsePos+trueNeg);
            println("	TPR	= "+tpr);
            println("	FPR	= "+fpr);
            println("");
        } catch (Exception e) {
            System.err.println("STATS Error: "+e);
        }
    }
    protected static void preproc(String dir,String filename,String tail) throws IOException {
        Scanner fin = new Scanner(new FileInputStream(dir+filename));
        PrintWriter fout = new PrintWriter(new FileOutputStream(dir+"new_"+filename));
        int cnt=0, total=0;
        while (fin.hasNextLine()) {
            StringTokenizer tok = new StringTokenizer(fin.nextLine());
            if (tok.nextToken().charAt(0)=='A') {
                fout.print("1 ");
                cnt++;
            } else {
                fout.print("-1 ");
            }
            while (tok.hasMoreTokens()) {
                fout.print(tok.nextToken()+" ");
            }
            fout.println(tail);
            total++;
        }
        System.out.println(filename+":	"+cnt+"/"+total);
        fout.close();
        fin.close();
    }
}


public class DIY extends AbstractDIY {
    private static int NUM;
    private static int NUM_OF_THREAD;
    
    private static Semaphore waitTask;
    private static Semaphore nextTask;
    private static Semaphore timeMutex;
    private static Subprob probSentinel;
    private static Augur modSentinel;
    
    private static BlockingQueue<Subprob> probBuf;
    private static BlockingQueue<Augur> modBuf;
    private static BlockingQueue<Integer> minIdx;
    
    private static AtomicInteger[][] minResult;
    
    static {
        try {
            init();
            waitTask = new Semaphore(0);
            nextTask = new Semaphore(0);
            timeMutex = new Semaphore(1);
            probSentinel = new Subprob(-1);
            modSentinel = new Augur();
            probBuf = new LinkedBlockingQueue<Subprob>();
            modBuf = new LinkedBlockingQueue<Augur>();
            minIdx = new LinkedBlockingQueue<Integer>();
        } catch (Exception e) {
            System.err.println("STATIC Error: "+e);
        }
    }
    private static void separate(List<Integer> poslst,List<Integer> neglst,boolean prior) {
        // List the positive samples and negative samples:
        // Precondition: poslst and neglst are non-null empty lists
        // Postcondition: poslst and neglst are filled with indices of positive
        //            and negative training data respectively
        if (!prior) {
            for (int i=0;i<train.l;i++) {
                if (train.getY(i)>0) {
                    poslst.add(new Integer(i));
                } else {
                    neglst.add(new Integer(i));
                }
            }
            Random rand = new Random();
            Collections.shuffle(poslst,rand);
            Collections.shuffle(neglst,rand);
        } else {
            try {
                System.out.println("Gathering Prior Knowledge ... ");
                BufferedReader pin = new BufferedReader(new FileReader("./data/train.txt"));
                List<DataItem> posData = new ArrayList<DataItem>();
                List<DataItem> negData = new ArrayList<DataItem>();
                for (int i=0;i<train.l;i++) {
                    String line = pin.readLine();
                    if (train.getY(i)>0) {
                        posData.add(new DataItem(i,line.substring(0,3)));
                    } else {
                        negData.add(new DataItem(i,line.substring(0,3)));
                    }
                }
                pin.close();
                //System.in.read();
                Collections.sort(posData);
                Collections.sort(negData);
                for (int i=0;i<posData.size();i++) {
                    poslst.add(posData.get(i).getValue());
                }
                for (int i=0;i<negData.size();i++) {
                    neglst.add(negData.get(i).getValue());
                }
                System.out.println("Ready!");
            } catch (Exception e) {
                System.err.println("SEPARATE Error: "+e);
            }
        }
    }
    private static void distribute(boolean prior) {
        List<Integer> poslst = new ArrayList<Integer>();
        List<Integer> neglst = new ArrayList<Integer>();
        separate(poslst,neglst,prior);
        // Group the positive samples and negative samples:
        int posGrpNum = (poslst.size()+NUM-1)/NUM;
        int negGrpNum = (neglst.size()+NUM-1)/NUM;
        int[] posGrps = new int[posGrpNum+1];
        int[] negGrps = new int[negGrpNum+1];
        System.out.println("	Totally "+posGrpNum*negGrpNum+" groups:");
        for (int i=0;i<posGrpNum;i++) {
            posGrps[i+1] = posGrps[i]+(poslst.size()+i)/posGrpNum;
        }
        for (int i=0;i<negGrpNum;i++) {
            negGrps[i+1] = negGrps[i]+(neglst.size()+i)/negGrpNum;
        }
        try {    // Add tasks to the buffer:
            for (int i=0;i<posGrpNum;i++) {
                for (int j=0;j<negGrpNum;j++) {
                    Subprob sub = new Subprob(i);
                    for (int k=posGrps[i];k<posGrps[i+1];k++) {
                        sub.add(poslst.get(k));
                    }
                    for (int k=negGrps[j];k<negGrps[j+1];k++) {
                        sub.add(neglst.get(k));
                    }
                    probBuf.put(sub);
                }
            }
        } catch (Exception e) {
            System.err.println("DISTRIBUTE Error 1: "+e);
        }
        try {    // Prepare for the MIN modules:
            minResult = new AtomicInteger[posGrpNum][test.l];
            for (int i=0;i<posGrpNum;i++) {
                for (int j=0;j<test.l;j++) {
                    minResult[i][j] = new AtomicInteger(1);
                }
            }
            probBuf.put(probSentinel);    // "STOP TRAINING" signal
        } catch (Exception e) {
            System.err.println("DISTRIBUTE Error 2: "+e);
        }
    }
    private static void main_thread(boolean prior) {
        try {
            start = System.currentTimeMillis();        // start training
            end = -1;                                // still pending
            DIY[] tsks = new DIY[NUM_OF_THREAD];
            for (int i=0;i<NUM_OF_THREAD;i++) {
                tsks[i] = new DIY();
                tsks[i].start();
            }
            distribute(prior);
            for (int i=0;i<NUM_OF_THREAD;i++) {
                waitTask.acquire();                    // awaiting sub-threads
            }
            for (int i=0;i<NUM_OF_THREAD;i++) {
                nextTask.release();                    // testing enabled
            }
            start = System.currentTimeMillis();        // start testing
            for (int i=0;i<NUM_OF_THREAD;i++) {
                tsks[i].join();
            }
            int[] res = new int[test.l];
            Arrays.fill(res,-1);
            for (int i=0;i<minResult.length;i++) {
                for (int j=0;j<test.l;j++) {
                    if (minResult[i][j].get()>0) {
                        res[j] = 1;                    // MAX Module
                    }
                }
            }
            end = System.currentTimeMillis();        // finish testing
            printTime(end-start,false);
            stats(res);
        } catch (Exception e) {
            System.err.println("MAIN_THRAD Error: "+e);
        }
    }
    public static void main(String[] args) {
        try {
            out = new PrintWriter(new FileWriter("./result/task"+args[0]+"_result.out"));
            NUM_OF_THREAD = Integer.parseInt(args[1]);
            int numMin = Integer.parseInt(args[2]);
            int numMax = Integer.parseInt(args[3]);
            int numStep = Integer.parseInt(args[4]);
            for (NUM=numMin;NUM<=numMax;NUM+=numStep) {
                print("Group Size = "+NUM+",	");
                println(NUM_OF_THREAD+" Threads");
                probBuf.clear();
                modBuf.clear();
                minIdx.clear();
                if (args[0].equals("6-2")) {
                    main_thread(false);
                } else {
                    main_thread(true);
                }
            }
            out.close();
        } catch (Exception e) {
            System.err.println("MAIN Error: "+e);
        }
    }
    
    public void run() {
        try {
            train();
            waitTask.release();
            nextTask.acquire();
            test();
        } catch (Exception e) {
            System.err.println("RUN Error: "+e);
        }
    }
    private void train() {
        Subprob sub = null;
        try {
            while (true) {
                sub = probBuf.take();
                if (sub==probSentinel) {    // signal of termination
                    timeMutex.acquire();
                    if (end<0) {            // finish training
                        end = System.currentTimeMillis();
                        printTime(end-start,true);
                    }
                    timeMutex.release();
                    probBuf.put(sub);
                    break;
                }
                DataSet prob = new DataSet(sub.size(),train.n);
                for (int i=0;i<sub.size();i++) {
                    int pos = sub.getItem(i).intValue();
                    prob.x.add(train.x.get(pos));
                    prob.y.add(train.y.get(pos));
                }
                modBuf.put(new Augur(prob));
                minIdx.put(sub.getIndex());
            }
            modBuf.put(modSentinel);
        } catch (Exception e) {
            System.err.println("TRAIN Error: "+e);
        }
    }
    private void test() {
        Augur model = null;
        int idx = -1;
        try {
            while (true) {
                model = modBuf.take();
                if (model==modSentinel) {    // signal of termination
                    modBuf.put(model);
                    break;
                }
                idx = minIdx.poll().intValue();
                for (int i=0;i<test.l;i++) {
                    if (model.predict(test.x.get(i))<0) {
                        minResult[idx][i].getAndSet(-1);    // MIN Modules
                    }
                }
            }
        } catch (Exception e) {
            System.err.println("TEST Error:"+e);
        }
    }
}

　　The final result is quite reassuring in terms of F₁ value and accuracy, whereas time performance leaves much to be desired: