20210607 TensorFlow 实现 Logistic 回归

Posted 2021-08-24 ATaburiss

0-1 导包

import warnings
warnings.filterwarnings("ignore")

import numpy as np
import tensorflow as tf
import matplotlib.pyplot as plt

1-1 构造数据

np.random.seed(999)
raw_data = np.random.standard_normal((1000, 2)) # 产生 0 均值的 正态分布

plt.figure(figsize=(8,6))       # 设置当前图像的长和宽
plt.scatter(raw_data[:,0],raw_data[:,1],marker=\'o\')
plt.grid()
plt.xlabel(\'x1\')
plt.ylabel(\'x2\')
plt.show()

1-1-2 现在定义一条直线，进行二分类

def pre(x):
    return -2.5 * x  - 0.5

# 横轴是 x1 纵轴是 x2
x1 = np.linspace(-2, 2, 500)    # 在 -2 到 2 之间，取出 500 个数
plt.figure(figsize=(8,6))
plt.plot(x1, pre(x1),\'salmon\')
plt.scatter(raw_data[:,0],raw_data[:,1],marker=\'o\')
plt.grid()
plt.xlabel(\'x1\')
plt.ylabel(\'x2\')
plt.show()

1-1-3 现在想区分数据
将x1的值代入直线方程，如果 x2 值，小于当前线上的值，就是直线下方的类

new_raw_data_0 = []
new_raw_data_1 = []
for one_data in raw_data:
    x1 = one_data[0]
    x2 = one_data[1]
    if x2 > pre(x1):
        tag = 1
        new_raw_data_1.append([x1,x2,tag])
    else:
        tag = 0
        new_raw_data_0.append([x1,x2,tag])

1-1-4 对二分类添加噪声

def add_noise(x):
    x = x + np.random.random(1)[0] - 0.5
    return x

new_raw_data_0 = [[add_noise(i[0]),add_noise(i[1]),i[2]] for i in new_raw_data_0]
new_raw_data_1 = [[add_noise(i[0]),add_noise(i[1]),i[2]] for i in new_raw_data_1]
new_raw_data_0 = np.array(new_raw_data_0)
new_raw_data_1 = np.array(new_raw_data_1)
# 列表生成式的方式添加噪声，然后变成 numpy 的数据格式

1-1-5 拿到数值后，画图

x1 = np.linspace(-2, 2, 500)
plt.figure(figsize=(8,6))
plt.plot(x1, pre(x1),\'salmon\')
plt.scatter(new_raw_data_0[:,0],new_raw_data_0[:,1],marker=\'o\')
plt.scatter(new_raw_data_1[:,0],new_raw_data_1[:,1],marker=\'o\')
plt.grid()
plt.xlabel(\'x1\')
plt.ylabel(\'x2\')
plt.show()

2-1 数据处理

# all data
all_data = np.concatenate([new_raw_data_0,new_raw_data_1])  # 拼接时，默认 axis = 0
# 现在，这样的数据是不可以放到神经网络中进行训练的
# 前面全是 0，后面全是 1，这样 loss 波动非常大，是不合理的
# 所以需要打乱顺序，使用 np.random.shuffle 将数据打乱，将数据丢进去，不需要返回值，all_data 会被修改，直接内部打乱
np.random.shuffle(all_data)  
# 然后将数据分出 训练集 和 测试集
train_data = all_data[:-64]
test_data = all_data[-64:]

2-2 数据分块

# 生成器
def gen_batch(data):
    np.random.shuffle(data)
    for i in range(len(data) // 64):
        cursor = 64 * i
        batch_data = data[cursor : cursor + 64]
        # 切块时，第一个 维度全要，第二个维度 有 3 个 数，取出前 2 个数
        x = batch_data[:, 0:2]  
        # y 只取 标签，0 或者 1
        y = batch_data[:, 2]
        yield x,y.reshape(-1,1)

2-2-1 运行生成器，查看结果

for x_,y_ in gen_batch(train_data):
    print(x_.shape)
    print(y_.shape)
    print(\'-------\')
    break

-->
(64, 2)
(64, 1)
-------
结果和预期一致

3-1 超参数

learing_rate = 0.01    # 梯度下降的学习率，一般都会传 0.01
num_train_epochs = 500  # 训练 500 次 
display_per_step = 100

3-2 计算图

graph = tf.Graph()  # 定义一个图
with graph.as_default():    # 默认图
    # None 代表第一维度不论是多少，都可以接收；分 64 块就是64，分 128 块就是128
    x = tf.placeholder(shape=[None,2], dtype=tf.float32, name=\'x\')
    y = tf.placeholder(shape=[None,1], dtype=tf.float32, name=\'y\')
    # w 设置了全 0，形状是(2,1)，因为 x1 x2 分别乘以 2 个 w
    # 当然，不建议全 0 初始化，全 0 初始化存在问题，
    # 因为这里的权重数量比较少，所以全0初始化问题不大
    w =  tf.Variable(tf.ones(shape=[2,1]), dtype=tf.float32)
    b =  tf.Variable(0, dtype=tf.float32)
    logits = tf.matmul(x, w) + b    # 矩阵运算，x乘以w, logits 是线性输出的结果
    # 逻辑回归，就是线性输出，套一个非线性函数，因为非线性函数可以把这些值映射到 0-1 之间
    # 可以以 0.5 作为边界，判断 0 或者 1
    y_pred = tf.sigmoid(logits)     # 将线性输出结果 logits 传进 非线性函数 sigmoid
    # 下一步，对y的真实值和预测值，做交叉熵
    # 定义loss，这里调用了 TensorFlow 的交叉熵函数，这里使用的是 sigmoid 交叉熵函数
    # tf.nn.sigmoid_cross_entropy_with_logits，传入了两个数，一个是标签，一个是 logits 线性输出结果
    # 当前的sigmoid交叉熵 会 自动进行 sigmoid 非线性变换，不需要 把已经变换完的 y_pred 传进来
    # 一定要传入 线性输出结果 logits，得到的交叉熵是 64 个结果，所以需要 reduce_mean，做平均
    loss = tf.reduce_mean(tf.nn.sigmoid_cross_entropy_with_logits(labels=y, logits=logits), name="calculate_loss")
    # 定义优化器
    optimizer = tf.train.GradientDescentOptimizer(learing_rate) 
    train_step = optimizer.minimize(loss)

3-3 运行计算图

with tf.Session(graph=graph) as sess:
    init = tf.global_variables_initializer()
    sess.run(init)
    step = 0
    for epoch in range(num_train_epochs):   
        for x_, y_ in gen_batch(train_data):
            step += 1
            _, l = sess.run([train_step, loss], feed_dict={x: x_, y: y_})
            if step % display_per_step == 0:
                print("step: {:>4}, loss: {:.4}".format(step, l))
    print(\'training over\')
    # 下面的是 测试集上的 loss
    x_test,y_test = next(gen_batch(test_data))
    loss_test = sess.run(loss,feed_dict={x: x_test, y: y_test})
    print("test loss is {:.4}".format(loss_test))
    res_weights = sess.run([w, b])
    print(res_weights)

-->
step:  100, loss: 0.3349
……
step: 7000, loss: 0.2156
training over
test loss is 0.2402
[array([[3.6807547],
       [1.3876256]], dtype=float32), 0.7486744]
# loss 是在逐步下降中，这里的w1 w2 和 b 值 和 前面的 wx+b的w和b 是不一样的
# 二分类后，可以添加评价指标，之前只看 loss，现在可以看除 loss 之外的 其他东西
# 0 1 标签是可以计算正确率的，线性回归中不涉及，因为线性回归中只是系数近似

4-1 添加正确率指标

graph = tf.Graph()
with graph.as_default():
    x = tf.placeholder(shape=[None,2], dtype=tf.float32, name=\'x\')
    y = tf.placeholder(shape=[None,1], dtype=tf.float32, name=\'y\')
    w =  tf.Variable(tf.ones(shape=[2,1]), dtype=tf.float32)
    b =  tf.Variable(0, dtype=tf.float32)
    logits = tf.matmul(x, w) + b
    y_pred = tf.sigmoid(logits)    
    # 命名空间，目的是收集权重；
    # 把某一块放到命名空间下，操作比较方便，计算图看起来更结构化一些
    # 定义loss
    with tf.name_scope("loss"):     
        loss = tf.reduce_mean(tf.nn.sigmoid_cross_entropy_with_logits(labels=y, logits=logits), name="calculate_loss")
    # 定义优化器
    with tf.name_scope("SGD"):
        optimizer = tf.train.GradientDescentOptimizer(learing_rate)
        train_step = optimizer.minimize(loss)
    # 定义正确率
    with tf.name_scope("calculate_accuracy"):
        res_pred = tf.cast(tf.greater_equal(y_pred, 0.5), dtype=tf.float32)     
        # 判断序列中的值 是否大于等于  0.5，是 返回 True，否 返回 False；tf.cast 强制转换成 0 1
        # 所以 res_pred 是一堆 0 1 值
        acc = tf.reduce_mean(tf.cast(tf.equal(res_pred, y), dtype=tf.float32))
        # 对应位置 一一判断 是否相等，对应下标的每一个 1 0 是否相等
with tf.Session(graph=graph) as sess:
    init = tf.global_variables_initializer()
    sess.run(init)
    step = 0
    for epoch in range(num_train_epochs):
        for x_, y_ in gen_batch(train_data):
            step += 1
            _, l, acc_ = sess.run([train_step, loss, acc], feed_dict={x: x_, y: y_})
            if step % display_per_step == 0:
                print("step: {:>4}, loss: {:.4}, acc: {:.4%}".format(step, l, acc_))
    print(\'training over\')
    x_test,y_test = next(gen_batch(test_data))
    loss_test, acc_test = sess.run([loss, acc],feed_dict={x: x_test, y: y_test})
    print("test loss is {:.4}, acc is {:.4%}".format(loss_test, acc_test))
    res_weights = sess.run([w, b])
    print(res_weights)

-->
step:  100, loss: 0.3412, acc: 85.9375%
step:  200, loss: 0.3945, acc: 82.8125%
……
step: 6900, loss: 0.2757, acc: 87.5000%
step: 7000, loss: 0.1816, acc: 93.7500%
training over
test loss is 0.2401, acc is 90.6250%
[array([[3.6801553],
       [1.3892566]], dtype=float32), 0.7487771]
# 正确率在震荡中，不断上升；测试集的 loss 是0.24，正确率约 92%
# 现在 逻辑回归 基本上就训练完了

5-1 画出二分类的线
# 思考过程
# 正向过程
# y_pred = sigmoid(w1*x1+w2*x2+b)
# if y_pred>=0.5 res_pred=1.0
# if y_pred<0.5 res_pred=0.0
# w1*x1+w2*x2+b = 0   y_pred 等于 0.5 的那条线；这是分界线
# x2 = (-w1*x1-b)/w2  以 x2 作为 y 值，所以画这条线即可
# w1*x1+w2*x2+b >= 0
# x2 >= (-w1*x1-b)/w2

def formula(x1, w1, w2, b):
    x2 = (-w1*x1-b)/w2
    return x2
plt.figure(figsize=(8,6))
line_space = np.linspace(-2, 2, 1024)
plt.scatter(new_raw_data_0[:,0],new_raw_data_0[:,1],marker=\'o\')
plt.scatter(new_raw_data_1[:,0],new_raw_data_1[:,1],marker=\'o\')
w1 = res_weights[0][0][0]
w2 = res_weights[0][1][0]
b = res_weights[1]
plt.plot(line_space, formula(line_space, w1, w2, b))
plt.grid()
plt.show()

部分理论说明：
https://blog.51cto.com/u_15149862/2875921
1. 如何参数初始化？为什么不建议全 0 初始化？
2. 什么是 Logistic 回归？
3. 什么是激活函数？激活函数都有哪些？
4. 正确率如何计算？除了正确率还有什么评价指标？