1. 数据加载
my_dataset_dataloader.py
"""
目的是把之前的__next__功能抽离出来
新建一个类，里面放__next__方法
"""
import random
import numpy as np
class MyDataset:
def __init__(self, all_datas, batch_size, shuffle=True):
self.all_datas = all_datas
self.batch_size = batch_size
self.shuffle = shuffle
# 在for循环中首次执行
def __iter__(self): # 返回一个具有__next__的对象
return Dataloader(self) # 把self作为dataset传入对象中
def __len__(self):
return len(self.all_datas)
class Dataloader:
def __init__(self, dataset):
self.dataset = dataset
# 根据长度生成一个索引，shuffle是打乱索引，而不是list重新排序
self.index = [i for i in range(len(self.dataset))]
if self.dataset.shuffle:
np.random.shuffle(self.index)
self.cursor = 0
def __next__(self):
if self.cursor >= len(self.dataset.all_datas):
raise StopIteration
# 将索引拿出来
index = self.index[self.cursor: self.cursor + self.dataset.batch_size]
# 根据索引取batch_data
batch_data = self.dataset.all_datas[index]
self.cursor += self.dataset.batch_size
return batch_data
if __name__ == "__main__":
all_datas = np.array([1, 2, 3, 4, 5, 6, 7])
batch_size = 2
shuffle = True
epoc = 2
dataset = MyDataset(all_datas, batch_size, shuffle)
for e in range(epoc):
for batch_data in dataset: # 把一个对象放在for上面，会自动调用这个对象的__iter__方法（仅仅首次执行）
# print(type(batch_data))
print(batch_data)
2. 多特征 - 线性回归
"""
线性回归的矩阵实现，即多个属性用矩阵表示
"""
import numpy as np
import pandas as pd
from tqdm import tqdm
import pickle
def get_data(file = "上海二手房价.csv"):
datas = pd.read_csv(file, names=['y', 'x1', 'x2', 'x3', 'x4', 'x5', 'x6'], skiprows=1)
y = datas['y'].values.reshape(-1, 1) # 变成纵向的向量
X = datas[[f'x{i}' for i in range(1, 7)]].values
# 归一化 z-score: (X - mean_x) / std
mean_y = np.mean(y)
std_y = np.std(y)
mean_X = np.mean(X, axis=0, keepdims=True)
std_X = np.std(X, axis=0, keepdims=True)
y = (y - mean_y) / std_y
X = (X - mean_X) / std_X
return X, y, mean_y, std_y, mean_X, std_X
if __name__ == '__main__':
X, y, mean_y, std_y, mean_X, std_X = get_data()
K = np.random.random((6, 1)) # 生成k的初始值，长度为6的列向量
epoc = 1000
"""
梯度爆炸的原因：
"""
lr = 0.05
b = 1
for e in range(epoc):
pre = X @ K + b
loss = np.sum((pre - y) ** 2) / len(X)
# 梯度计算部分
# G = 2 * (pre - y) # 是一个长度为6的向量
G = (pre - y) / len(X) # 去掉梯度的常数项，并且除以长度，防止梯度爆照
delta_K = X.T @ G
delta_b = np.sum(G) # 偏置项的梯度就是G
K = K - lr * delta_K # 迭代Ｋ
b = b - lr * delta_b
print(loss)
# while True:
# bedroom = (int(input('请输入卧室数量：')))
# ting = (int(input('请输入客厅餐厅数量：')))
# wei = (int(input('请输入卫生间数量：')))
# area = (int(input('请输入面积：')))
# floor = (int(input('请输入楼层：')))
# year = (int(input('请输入建成年份：')))
#
# text_x = (np.array([bedroom, ting, wei, area, floor, year]).reshape(1, -1) - mean_X) / std_X
#
# p = text_x @ K
# print(f'房价为{p + std_y + mean_y}')3. 逻辑回归
"""
激活函数，目的是把pre映射到一个目标区间内
"""
import numpy as np
def sigmoid(x):
return 1 / (1 + np.exp(-x))
if __name__ == '__main__':
# ---------------毛发长，腿长
dogs = np.array([[8.9, 12], [9, 11], [10, 13], [9.9, 11.2], [12.2, 10.1], [9.8, 13], [8.8, 11.2]]) # 0
cats = np.array([[3, 4], [5, 6], [3.5, 5.5], [4.5, 5.1], [3.4, 4.1], [4.1, 5.2], [4.4, 4.4]]) # 1
labels = np.array([0] * len(dogs) + [1] * len(cats), dtype=np.int32).reshape(-1, 1)
# X = np.concatenate((dogs, cats), axis=0)
X = np.vstack((dogs, cats)) # 纵向合并
K = np.random.normal(0, 1, size=(2, 1))
b = 0
epoc = 1000
lr = 0.05
for e in range(epoc):
p = X @ K + b
pre = sigmoid(p)
# log中的值是0~1的值，所以结果是负数，加负号
loss = -np.mean(labels * np.log(pre) + (1 - labels) * np.log(1 - pre))
G = pre - labels # G为loss对pre的导数
delta_K = X.T @ G
delta_b = np.sum(G)
K -= lr * delta_K
b -= lr * delta_b
# print(loss)
while True:
f1 = float(input('请输入毛发长：'))
f2 = float(input('请输入腿长：'))
text_x = np.array([f1, f2]).reshape(1, 2)
p = sigmoid(text_x @ K + b)
if p > 0.5:
print('类别：猫')
else:
print('类别：狗')4. BP神经网络
import numpy as np
import struct
import matplotlib.pyplot as plt
def load_labels(file): # 加载标签
with open(file, 'rb') as f:
data = f.read()
return np.asanyarray(bytearray(data[8:]), dtype=np.int32)
def load_images(file): # 加载图片
with open(file, 'rb') as f:
data = f.read()
magic_number, num_items, rows, cols = struct.unpack('>iiii', data[: 16])
return np.asanyarray(bytearray(data[16:]), dtype=np.uint8).reshape(num_items, -1)
def make_one_hot(labels, class_num=10): # 独热编码
result = np.zeros((len(labels), class_num))
for index, label in enumerate(labels):
result[index][label] = 1
return result
def sigmoid(x):
return 1 / (1 + np.exp(-x))
def softmax(x):
return np.exp(x) / np.sum(np.exp(x), axis=1, keepdims=True)
if __name__ == '__main__':
train_datas = load_images(r'data\train-images.idx3-ubyte') / 255 # 数据归一化，图片像素大小是0~255
train_labels = load_labels(r'data\train-labels.idx1-ubyte')
train_labels = make_one_hot(train_labels) # 变化为独热码
test_datas = load_images(r'data\t10k-images.idx3-ubyte') / 255
test_labels = load_labels(r'data\t10k-labels.idx1-ubyte')
epoc = 20
batch_size = 200
lr = 0.1
hidden_num = 256
w1 = np.random.normal(0, 1, size=(train_datas.shape[1], hidden_num))
w2 = np.random.normal(0, 1, size=(hidden_num, 10))
batch_times = int(np.ceil(len(train_datas) / batch_size)) # 向上取整，然后化为整数
for e in range(epoc):
for batch_index in range(batch_times):
# ---------- 获取数据 -------------
batch_x = train_datas[batch_index * batch_size: (batch_index + 1) * batch_size]
batch_label = train_labels[batch_index * batch_size: (batch_index + 1) * batch_size]
# ---------- forward -------------
h = batch_x @ w1
sig_h = sigmoid(h)
p = sig_h @ w2
pre = softmax(p)
# ---------- 计算loss -------------
loss = -np.sum(batch_label * np.log(pre)) / batch_size # 这里是计算的平均loss
# ---------- backward -------------
"""
当batch_size很大的时候，G2会带动下面所有的数值，导致梯度爆炸
理论上，batch_size越大，数据运行越快，每个epoc下降速度会越慢
"""
G2 = (pre - batch_label) / batch_size # 防止因batch_size太大时候导致的梯度爆炸
delta_w2 = sig_h.T @ G2
delta_sig_h = G2 @ w2.T
delta_h = delta_sig_h * sig_h * (1 - sig_h)
delta_w1 = batch_x.T @ delta_h
# ---------- 更新梯度 -------------
w1 -= lr * delta_w1
w2 -= lr * delta_w2
# 计算精确度
h = test_datas @ w1
sig_h = sigmoid(h)
p = sig_h @ w2
pre = softmax(p)
pre = np.argmax(pre, axis=1)
acc = np.sum(pre == test_labels) / 10000
print(acc)5. 卷积
import cv2
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import os
def get_imgs(path:str):
# 读入图片
img_files = os.listdir(path)
result = []
for file in img_files:
file = os.path.join(path, file)
img = cv2.imread(file)
img = cv2.resize(img, (150, 150))
img = img.transpose(2, 0, 1)
result.append(img)
return np.array(result)
def conv(imgs, kernel):
out_channel, in_channel, kernel_h, kernel_w = kernel.shape
img_num, _, img_h, img_w = imgs.shape
# 卷积之后的宽和高
c_w = img_w - kernel_w + 1 # width
c_h = img_h - kernel_h + 1 # high
A = kernel.reshape(out_channel, -1)
B = np.zeros((img_num, A.shape[1], c_w * c_h))
c = 0
for h in range(c_h): # 后从上到下
for w in range(c_w): # 先从左到右
d = imgs[:, :, h: h+kernel_h, w: w+kernel_w]
d = d.reshape(img_num, -1) # 假如是(1, -1)则会变成二维的，与下面切片出来的一维不匹配
B[:, :, c] = d
c += 1
result = A @ B
result = result.reshape(img_num, out_channel, c_h, c_w)
return result
if __name__ == '__main__':
kernel = np.array(
[
[
[
[-1, -2, -3],
[-1, -2, -3],
[-1, -10, 1]
],
[
[0, 3, 3],
[-1, -2, -3],
[1, 1, 1]
],
[
[3, 3, 3],
[-1, -9, -3],
[1, 1, 1]
]
],
[
[
[1, -1, 0],
[1, -1, 0],
[1, -1, 0]
],
[
[1, -1, 0],
[1, -1, 0],
[1, -1, 0]
],
[
[1, -1, 0],
[1, -1, 0],
[1, -1, 0]
]
]
]
)
imgs = get_imgs('img')
result = conv(imgs, kernel)
for i in result:
for j in i:
plt.imshow(j, cmap='gray')
plt.show()6. 封装
import numpy as np
import struct
import matplotlib.pyplot as plt
def load_labels(file): # 加载标签
with open(file, 'rb') as f:
data = f.read()
return np.asanyarray(bytearray(data[8:]), dtype=np.int32)
def load_images(file): # 加载图片
with open(file, 'rb') as f:
data = f.read()
magic_number, num_items, rows, cols = struct.unpack('>iiii', data[: 16])
return np.asanyarray(bytearray(data[16:]), dtype=np.uint8).reshape(num_items, -1)
def make_one_hot(labels, class_num=10): # 独热编码
result = np.zeros((len(labels), class_num))
for index, label in enumerate(labels):
result[index][label] = 1
return result
def sigmoid(x):
return 1 / (1 + np.exp(-x))
def softmax(x):
return np.exp(x) / np.sum(np.exp(x), axis=1, keepdims=True)
def get_datas():
train_datas = load_images(r'..\05_01_05BP神经网络\data\train-images.idx3-ubyte') / 255 # 数据归一化，图片像素大小是0~255
train_labels = load_labels(r'..\05_01_05BP神经网络\data\train-labels.idx1-ubyte')
train_labels = make_one_hot(train_labels) # 变化为独热码
test_datas = load_images(r'..\05_01_05BP神经网络\data\t10k-images.idx3-ubyte') / 255
test_labels = load_labels(r'..\05_01_05BP神经网络\data\t10k-labels.idx1-ubyte')
return train_datas, train_labels, test_datas, test_labels
class Linear: # 用linear层替换w1和w2
def __init__(self, in_num, out_num):
self.weight = np.random.normal(0, 1, size=(in_num, out_num))
def __call__(self, x): # 对象+括号 就会调用__call__函数
return self.forward(x)
def forward(self, x):
self.x = x
return x @ self.weight
def backward(self, G):
delta_weight = self.x.T @ G
self.weight -= lr * delta_weight # 优化器的内容，梯度下降优化器SGD
delta_x = G @ self.weight.T
return delta_x
class Sigmoid:
def __call__(self, x):
return self.forward(x)
def forward(self, x):
self.r = sigmoid(x)
return self.r
def backward(self, G):
return G * self.r * (1 - self.r)
class Softmax:
def __call__(self, x):
return self.forward(x)
def forward(self, x):
self.r = softmax(x)
return self.r
def backward(self, G): # G传入label的值
return (self.r - G) / self.r.shape[0]
class Mymodel:
def __init__(self, layers):
self.layers = layers
def forward(self, x, label=None):
for layer in self.layers:
x = layer(x)
self.x = x
if label is not None:
self.label = label
loss = -np.sum(label * np.log(x)) / x.shape[0]
return loss
def backward(self):
G = self.label
for layer in self.layers[::-1]:
G = layer.backward(G)
def __call__(self, *args): # *args表示未命名变参，**kwargs表示命名变参
return self.forward(*args)
if __name__ == '__main__':
# 获取数据
train_datas, train_labels, test_datas, test_labels = get_datas()
epoch = 20
batch_size = 200
lr = 0.1
hidden_num = 256
model = Mymodel(
[
Linear(train_datas.shape[1], hidden_num),
Sigmoid(),
Linear(hidden_num, 10),
Softmax()
]
)
batch_times = int(np.ceil(len(train_datas) / batch_size)) # 向上取整，然后化为整数
for e in range(epoch):
for batch_index in range(batch_times):
# ---------- 获取数据 -------------
x = train_datas[batch_index * batch_size: (batch_index + 1) * batch_size]
batch_label = train_labels[batch_index * batch_size: (batch_index + 1) * batch_size]
# ---------- forward -------------
loss = model(x, batch_label)
if batch_index % 100 == 0:
print(f'loss: {loss:.3f}')
# ---------- backward -------------
model.backward()
# 计算精确度
x = test_datas
model(x)
pre = np.argmax(model.x, axis=1)
acc = np.sum(pre == test_labels) / 10000
print(f'--------------acc: {acc:.3f}')