NumPy 矩阵计算与数据实战

import numpy as np

# 创建数组
arr = np.array([[1, 2], [3, 4]])
print(f"数组类型: {type(arr)}")       # <class 'numpy.ndarray'>
print(f"数组 * 数组 (逐元素):\n{arr * arr}")
# 输出:
# [[ 1  4]
#  [ 9 16]]

# 创建矩阵
mat = np.mat([[1, 2], [3, 4]])
print(f"\n矩阵类型: {type(mat)}")       # <class 'numpy.matrix'>
print(f"矩阵 * 矩阵 (矩阵乘法):\n{mat * mat}")
# 输出:
# [[ 7 10]
#  [15 22]]

import numpy as np

# 从数组转换为矩阵
arr = np.array([[1, 2], [3, 4]])
mat = np.mat(arr)

# 从矩阵转回数组
arr_back = np.asarray(mat)
print(f"转换后类型: {type(arr_back)}")  # <class 'numpy.ndarray'>

import numpy as np

A = np.array([[1, 2], [3, 4]])
B = np.array([[5, 6], [7, 8]])

# 形状相同的加减法
print("A + B:")
print(A + B)   # [[ 6  8]
               # [10 12]]

# 广播：标量加矩阵
print("\nA + 10:")
print(A + 10)  # [[11 12]
               # [13 14]]

A = np.array([[1, 2], [3, 4]])

# 数乘：每个元素乘以标量
print("3 * A:")
print(3 * A)   # [[ 3  6]
               # [ 9 12]]

print("\nA * 3:")
print(A * 3)   # 结果相同

import numpy as np

# 一维数组 → 二维矩阵
arr_1d = np.array([1, 2, 3, 4, 5, 6])

# reshape 为 2×3 矩阵
mat_2d = arr_1d.reshape(2, 3)
print(f"reshape(2, 3):\n{mat_2d}")
# [[1 2 3]
#  [4 5 6]]

# reshape 为 3×2 矩阵
mat_2x3 = arr_1d.reshape(3, 2)
print(f"\nreshape(3, 2):\n{mat_2x3}")
# [[1 2]
#  [3 4]
#  [5 6]]

# 使用 -1 自动推断维度
auto_shape = arr_1d.reshape(2, -1)  # -1 表示自动计算
print(f"\nreshape(2, -1):\n{auto_shape}")  # 同 reshape(2, 3)

import numpy as np

arr = np.array([1, 2, 3, 4])  # shape: (4,)

# 增加行维度 → 变成列向量 (4, 1)
col_vec = arr[:, np.newaxis]
print(f"列向量 shape: {col_vec.shape}")  # (4, 1)
print(col_vec)
# [[1]
#  [2]
#  [3]
#  [4]]

# 增加列维度 → 变成行向量 (1, 4)
row_vec = arr[np.newaxis, :]
print(f"\n行向量 shape: {row_vec.shape}")  # (1, 4)
print(row_vec)  # [[1 2 3 4]]

import numpy as np

A = np.array([[1, 2], [3, 4]])   # 2×2 矩阵
B = np.array([[5, 6], [7, 8]])   # 2×2 矩阵

# 方式一：@ 运算符（推荐）
result1 = A @ B
print("@ 运算符:\n", result1)
# [[19 22]
#  [43 50]]

# 方式二：np.dot()
result2 = np.dot(A, B)
print("\nnp.dot():\n", result2)  # 结果相同

# 方式三：np.matmul()
result3 = np.matmul(A, B)
print("\nnp.matmul():\n", result3)  # 结果相同

# 验证形状规则
A = np.random.randn(3, 4)   # 3×4
B = np.random.randn(4, 5)   # 4×5
C = A @ B                    # 结果: 3×5
print(f"A: {A.shape}, B: {B.shape}, A@B: {C.shape}")
# A: (3, 4), B: (4, 5), A@B: (3, 5)

import numpy as np

# 功能一：向量点积（一维输入）
v1 = np.array([1, 2, 3])
v2 = np.array([4, 5, 6])
dot_product = np.dot(v1, v2)
print(f"向量点积: {dot_product}")  # 32 (= 1×4 + 2×5 + 3×6)

# 功能二：矩阵乘法（二维输入）
A = np.array([[1, 2], [3, 4]])
B = np.array([[5, 6], [7, 8]])
matrix_product = np.dot(A, B)
print(f"矩阵乘法:\n{matrix_product}")
# [[19 22]
#  [43 50]]

import numpy as np

A = np.array([[1, 2, 3], [4, 5, 6]])

# 转置
print("转置 A.T:")
print(A.T)
# [[1 4]
#  [2 5]
#  [3 6]]

# 展平
print(f"\n展平: {A.flatten()}")  # [1 2 3 4 5 6]

# squeeze 示例
arr = np.array([[[1, 2, 3]]])  # shape: (1, 1, 3)
squeezed = np.squeeze(arr)
print(f"squeeze 前: {arr.shape}, 后: {squeezed.shape}")
# squeeze 前: (1, 1, 3), 后: (3,)

import numpy as np

# 准备数据
a = np.array([[1, 2], [3, 4]])     # 2×2
b = np.array([[5, 6], [7, 8]])     # 2×2

# 方式一：concatenate(axis=1)
result1 = np.concatenate([a, b], axis=1)
print("concatenate(axis=1):\n", result1)
# [[1 2 5 6]
#  [3 4 7 8]]

# 方式二：hstack
result2 = np.hstack([a, b])
print("\nhstack:\n", result2)  # 结果相同

# 方式三：column_stack（自动处理一维数组）
c = np.array([9, 10])          # 一维数组
result3 = np.column_stack([a, c])
print("\ncolumn_stack (含一维):\n", result3)
# [[ 1  2  9]
#  [ 3  4 10]]

import numpy as np

a = np.array([[1, 2], [3, 4]])     # 2×2
b = np.array([[5, 6], [7, 8]])     # 2×2

# 垂直合并
result = np.vstack([a, b])
print("vstack:\n", result)
# [[1 2]
#  [3 4]
#  [5 6]
#  [7 8]]

# row_stack 与 vstack 完全相同
result2 = np.row_stack([a, b])
print("\nvstack == row_stack:", np.array_equal(result, result2))  # True

# ❌ 错误示例：行数不同无法水平合并
a = np.array([[1, 2], [3, 4]])        # 2 行
b = np.array([[5, 6], [7, 8], [9, 10]])  # 3 行
try:
    result = np.hstack([a, b])         # 报错！
except ValueError as e:
    print(f"错误: {e}")
    # all input arrays must have the same shape...

import numpy as np

arr = np.arange(12).reshape(4, 3)
print("原始数组:\n", arr)
# [[ 0  1  2]
#  [ 3  4  5]
#  [ 6  7  8]
#  [ 9 10 11]]

# 垂直等分为 2 份
parts = np.split(arr, 2, axis=0)
print(f"\n垂直等分 2 份: {len(parts)} 个数组")
for i, part in enumerate(parts):
    print(f"  第 {i} 份:\n{part}")

# 水平等分为 3 份
parts_h = np.split(arr, 3, axis=1)
print(f"\n水平等分 3 份: {len(parts_h)} 个数组")
for i, part in enumerate(parts_h):
    print(f"  第 {i} 份:\n{part}")

import numpy as np

arr = np.arange(10)
print(f"原始数组: {arr}")  # [0 1 2 3 4 5 6 7 8 9]

# 在索引 3 和 7 处分割 → 得到 3 份数组
parts = np.split(arr, [3, 7])
print(f"\n按 [3, 7] 分割: {len(parts)} 份")
for i, part in enumerate(parts):
    print(f"  第 {i} 份: {part}")
# 第 0 份: [0 1 2]       （索引 0-2）
# 第 1 份: [3 4 5 6]     （索引 3-6）
# 第 2 份: [7 8 9]       （索引 7-9）

# 相关函数速查
# np.array_split()  — 不等分也允许（当不能整除时）
# np.vsplit()      — 垂直分割的简写（axis=0）
# np.hsplit()      — 水平分割的简写（axis=1）

import numpy as np

# 模拟：从三个数据源读取的数据
data_jan = np.array([
    [100, 1500],
    [102, 1600],
    [98, 1400],
])  # 1 月数据: [价格, 成交量]

data_feb = np.array([
    [105, 1700],
    [108, 1800],
    [103, 1550],
    [110, 1900],
])  # 2 月数据

data_mar = np.array([
    [112, 2000],
    [109, 1850],
])  # 3 月数据

# 步骤 1：垂直合并（按时间顺序堆叠）
all_data = np.vstack([data_jan, data_feb, data_mar])
print(f"合并后 shape: {all_data.shape}")  # (9, 2)

# 步骤 2：添加日期标签列
dates = np.array([
    ['2026-01-01'], ['2026-01-02'], ['2026-01-03'],
    ['2026-02-01'], ['2026-02-02'], ['2026-02-03'], ['2026-02-04'],
    ['2026-03-01'], ['2026-03-02'],
])

# 水平拼接日期列
final_data = np.column_stack([dates, all_data])
print(f"\n最终数据 ({final_data.shape}):")
print(final_data[:5])  # 显示前 5 行

import numpy as np

# ====== loadtxt: 快速读取干净数据 ======
# 假设 stock.csv 格式：
# date,open,high,low,close,volume
# 2026-01-02,100.5,102.0,99.0,101.0,1500000
# ...

# 只读取数值列（跳过首行 header，指定列）
data_loadtxt = np.loadtxt(
    'stock.csv',
    delimiter=',',
    skiprows=1,
    usecols=(1, 2, 3, 4, 5),  # 读取 open, high, low, close, volume
)
print(f"loadtxt 读取: shape={data_loadtxt.shape}, dtype={data_loadtxt.dtype}")

# ====== genfromtxt: 处理含缺失值数据 ======
# genfromtxt 会将缺失值填充为 nan
data_gen = np.genfromtxt(
    'stock_dirty.csv',
    delimiter=',',
    skiprows=1,
    usecols=(1, 2, 3, 4, 5),
    missing_values='',           # 指定什么算缺失值
    filling_values=np.nan,       # 缺失值用什么填充
)
print(f"\ngenfromtxt 读取: shape={data_gen.shape}")
print(f"缺失值数量: {np.isnan(data_gen).sum()}")

import numpy as np

# 假设已读取数据: 每行 [open, high, low, close, volume]
# 这里用模拟数据演示
np.random.seed(42)
n_days = 60
base_price = 100

open_prices = base_price + np.cumsum(np.random.randn(n_days) * 0.5)
high_prices = open_prices + np.abs(np.random.randn(n_days) * 1.5)
low_prices = open_prices - np.abs(np.random.randn(n_days) * 1.5)
close_prices = open_prices + np.random.randn(n_days) * 1.0
volumes = np.random.randint(1000000, 5000000, size=n_days).astype(float)

# 组装为数据矩阵: (60, 5)
stock_data = np.column_stack([open_prices, high_prices, low_prices, close_prices, volumes])

# ---- 基本信息探索 ----
print("=" * 50)
print("数据探索报告")
print("=" * 50)
print(f"数据形状: {stock_data.shape}")              # (60, 5) → 60 天, 5 个字段
print(f"数据类型: {stock_data.dtype}")              # float64
print(f"内存占用: {stock_data.nbytes / 1024:.1f} KB")

# 各列含义
columns = ['开盘价', '最高价', '最低价', '收盘价', '成交量']
print(f"\n各列: {columns}")

# ---- 切片获取子集 ----
# 获取前 5 天的数据
first_5_days = stock_data[:5, :]
print(f"\n前 5 天数据:\n{first_5_days}")

# 只获取收盘价（第 4 列，索引 3）
close = stock_data[:, 3]
print(f"\n收盘价序列 (前 10 个): {close[:10].round(2)}")

# 获取最后 10 天的开盘价和收盘价
recent_oc = stock_data[-10:, [0, 3]]  # 第 0 列和第 3 列
print(f"\n最近 10 天开盘价和收盘价:\n{recent_oc.round(2)}")

def min_max_normalize(data):
    """Min-Max 标准化"""
    min_val = data.min(axis=0)
    max_val = data.max(axis=0)
    return (data - min_val) / (max_val - min_val)

# 对除成交量外的价格数据进行归一化
price_data = stock_data[:, :4]  # 取前 4 列（价格相关）
normalized_prices = min_max_normalize(price_data)

print("Min-Max 归一化结果 (前 5 行):")
print(normalized_prices[:5].round(3))
# 每列都在 [0, 1] 区间内

def z_score_normalize(data):
    """Z-score 标准化"""
    mean = data.mean(axis=0)
    std = data.std(axis=0)
    return (data - mean) / std

normalized_z = z_score_normalize(price_data)

print("\nZ-score 归一化结果验证:")
print(f"每列均值:\n{normalized_z.mean(axis=0).round(6)}")  # ≈ 0
print(f"每列标准差:\n{normalized_z.std(axis=0).round(6)}")  # ≈ 1

import numpy as np

close = stock_data[:, 3]  # 收盘价
volume = stock_data[:, 4]  # 成交量

print("=" * 50)
print("描述性统计分析")
print("=" * 50)

# ---- 基本统计量 ----
print(f"\n【收盘价统计】")
print(f"  均值:     {close.mean():.2f}")
print(f"  中位数:   {np.median(close):.2f}")
print(f"  标准差:   {close.std():.2f}")
print(f"  最大值:   {close.max():.2f}")
print(f"  最小值:   {close.min():.2f}")
print(f"  极差:     {close.max() - close.min():.2f}")

print(f"\n【成交量统计】")
print(f"  均值:     {volume.mean():,.0f}")
print(f"  标准差:   {volume.std():,.0f}")
print(f"  总成交量: {volume.sum():,.0f}")

# ---- 稳定性判断 ----
cv = close.std() / close.mean()  # 变异系数 Coefficient of Variation
print(f"\n【稳定性评估】")
print(f"  变异系数 CV: {cv:.4f}")
if cv < 0.05:
    print("  → 波动较小，相对稳定")
elif cv < 0.15:
    print("  → 波动适中")
else:
    print("  → 波动较大，风险较高")

# ---- 各价格列统计摘要 ----
price_columns = ['开盘', '最高', '最低', '收盘']
print(f"\n【各价格列统计】")
print(f"{'字段':<8} {'均值':>10} {'标准差':>10} {'最小':>10} {'最大':>10}")
print("-" * 52)
for i, name in enumerate(price_columns):
    col = stock_data[:, i]
    print(f"{name:<8} {col.mean():>10.2f} {col.std():>10.2f} "
          f"{col.min():>10.2f} {col.max():>10.2f}")

import numpy as np

close = stock_data[:, 3]
volume = stock_data[:, 4]

# 计算 VWAP
vwap = np.average(close, weights=volume)
print(f"VWAP (成交量加权均价): {vwap:.2f}")

# 对比简单均价
simple_avg = close.mean()
print(f"简单算术均价: {simple_avg:.2f}")
print(f"差异: {vwap - simple_avg:+.2f} "
      f"({((vwap / simple_avg - 1) * 100):+.2f}%)")

# 解释：如果 VWAP > 简单均价，
# 说明高价位成交更多（资金追涨倾向）

open_p = stock_data[:, 0]
close_p = stock_data[:, 3]

# 日涨跌幅 (%)
daily_return = (close_p - open_p) / open_p * 100

print("日涨跌幅统计:")
print(f"  平均涨跌: {daily_return.mean():+.3f}%")
print(f"  最大涨幅: {daily_return.max():+.3f}%")
print(f"  最大跌幅: {daily_return.min():+.3f}%")

# 找出涨幅最大的一天
best_day = np.argmax(daily_return)
print(f"\n最佳交易日: 第 {best_day + 1} 天, 涨幅 {daily_return[best_day]:+.3f}%")

# 找出跌幅最大的一天
worst_day = np.argmin(daily_return)
print(f"最差交易日: 第 {worst_day + 1} 天, 跌幅 {daily_return[worst_day]:+.3f}%")

high = stock_data[:, 1]
low = stock_data[:, 2]
volume = stock_data[:, 4]

# 计算每日成交额
avg_price = (high + low) / 2
turnover = volume * avg_price

print(f"每日成交额 (万元):")
print(f"  总成交额: {turnover.sum() / 10000:,.1f} 万元")
print(f"  日均成交额: {turnover.mean() / 10000:,.1f} 万元")
print(f"  单日最高: {turnover.max() / 10000:,.1f} 万元 (第 {np.argmax(turnover)+1} 天)")
print(f"  单日最低: {turnover.min() / 10000:,.1f} 万元 (第 {np.argmin(turnover)+1} 天)")

import numpy as np

daily_return = (stock_data[:, 3] - stock_data[:, 0]) / stock_data[:, 0] * 100

# ---- 涨跌天数统计 ----
up_days = daily_return > 0
down_days = daily_return < 0
flat_days = daily_return == 0

print("涨跌天数统计:")
print(f"  上涨天数: {up_days.sum()} 天 ({up_days.mean()*100:.1f}%)")
print(f"  下跌天数: {down_days.sum()} 天 ({down_days.mean()*100:.1f}%)")
print(f"  平盘天数: {flat_days.sum()} 天")

# ---- 上涨日的平均涨幅 ----
up_returns = daily_return[up_days]
down_returns = daily_return[down_days]

print(f"\n上涨日分析:")
print(f"  平均涨幅: {up_returns.mean():+.3f}%")
print(f"  最大单日涨幅: {up_returns.max():+.3f}%")

print(f"\n下跌日分析:")
print(f"  平均跌幅: {down_returns.mean():+.3f}%")
print(f"  最大单日跌幅: {down_returns.min():+.3f}%")

# ---- 大幅波动筛选 ----
# 定义大幅波动为 |涨跌幅| > 2%
threshold = 2.0
volatile = np.abs(daily_return) > threshold

print(f"\n大幅波动 (|涨跌幅| > {threshold}%):")
print(f"  天数: {volatile.sum()} 天")
if volatile.sum() > 0:
    volatile_returns = daily_return[volatile]
    volatile_indices = np.where(volatile)[0]
    print(f"  详情:")
    for idx in volatile_indices:
        direction = "📈 涨" if daily_return[idx] > 0 else "📉 跌"
        print(f"    第 {idx+1} 天: {direction} {daily_return[idx]:+.3f}%")

# ---- 连续涨跌统计 ----
# 用 diff 找到符号变化的位置
sign = np.sign(daily_return)
sign[sign == 0] = 1  # 平盘视为正向
changes = np.diff(sign) != 0
change_points = np.where(changes)[0]

segments = np.split(daily_return, change_points + 1)
print(f"\n连续走势段数: {len(segments)}")
for i, seg in enumerate(segments[:5]):  # 只显示前 5 段
    trend = "连续上涨" if seg[0] > 0 else "连续下跌"
    print(f"  第{i+1}段: {trend} {len(seg)}天")

import numpy as np
import matplotlib.pyplot as plt

# 设置中文字体（macOS）
plt.rcParams['font.sans-serif'] = ['Arial Unicode MS', 'SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 准备数据
days = np.arange(1, len(stock_data) + 1)
close = stock_data[:, 3]
open_p = stock_data[:, 0]
daily_return = (close - open_p) / open_p * 100

# 创建图表布局
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
fig.suptitle('股票数据分析报告', fontsize=16, fontweight='bold')

# ---- 图 1：收盘价走势 ----
ax1 = axes[0, 0]
ax1.plot(days, close, 'b-', linewidth=1.5, label='收盘价')
ax1.plot(days, open_p, 'g--', linewidth=0.8, alpha=0.7, label='开盘价')
ax1.fill_between(days, close, open_p,
                  where=(close >= open_p), color='red', alpha=0.3, label='阳线')
ax1.fill_between(days, close, open_p,
                  where=(close < open_p), color='green', alpha=0.3, label='阴线')
ax1.set_xlabel('交易日')
ax1.set_ylabel('价格')
ax1.set_title('价格走势图')
ax1.legend(loc='upper left', fontsize=8)
ax1.grid(True, alpha=0.3)

# ---- 图 2：日涨跌幅柱状图 ----
ax2 = axes[0, 1]
colors = ['#ef5350' if r > 0 else '#26a69a' for r in daily_return]
ax2.bar(days, daily_return, color=colors, width=0.8, edgecolor='none', alpha=0.8)
ax2.axhline(y=0, color='black', linewidth=0.5)
ax2.axhline(y=daily_return.mean(), color='blue', linewidth=1,
            linestyle='--', label=f'均值: {daily_return.mean():+.2f}%')
ax2.set_xlabel('交易日')
ax2.set_ylabel('涨跌幅 (%)')
ax2.set_title('日涨跌幅分布')
ax2.legend(fontsize=8)
ax2.grid(True, alpha=0.3, axis='y')

# ---- 图 3：成交量 ----
ax3 = axes[1, 0]
volume = stock_data[:, 4]
vol_colors = ['#ef5350' if close[i] >= open_p[i] else '#26a69a'
              for i in range(len(close))]
ax3.bar(days, volume / 10000, color=vol_colors, width=0.8, alpha=0.7)
ax3.set_xlabel('交易日')
ax3.set_ylabel('成交量 (万手)')
ax3.set_title('成交量分布')
ax3.grid(True, alpha=0.3, axis='y')

# ---- 图 4：涨跌幅分布直方图 ----
ax4 = axes[1, 1]
ax4.hist(daily_return, bins=20, color='steelblue', edgecolor='white', alpha=0.8)
ax4.axvline(x=0, color='red', linewidth=1.5, linestyle='-')
ax4.axvline(x=daily_return.mean(), color='orange', linewidth=1.5,
            linestyle='--', label=f'均值: {daily_return.mean():+.2f}%')
ax4.axvline(x=daily_return.std(), color='green', linewidth=1,
            linestyle=':', label=f'+1σ: {daily_return.std():+.2f}%')
ax4.axvline(x=-daily_return.std(), color='green', linewidth=1,
            linestyle=':', label=f'-1σ: {-daily_return.std():+.2f}%')
ax4.set_xlabel('涨跌幅 (%)')
ax4.set_ylabel('频次')
ax4.set_title('涨跌幅分布直方图')
ax4.legend(fontsize=8)
ax4.grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('stock_analysis_report.png', dpi=150, bbox_inches='tight')
plt.show()

print("✅ 图表已保存: stock_analysis_report.png")

import numpy as np

# ❌ 慢：Python 循环
data = np.random.randn(1000000)
result = np.zeros_like(data)
for i in range(len(data)):
    result[i] = data[i] ** 2 + 2 * data[i]

# ✅ 快：向量化
result_fast = data ** 2 + 2 * data

# ✅ 更快：就地操作
result_inplace = np.empty_like(data)
np.multiply(data, data, out=result_inplace)   # out 参数避免临时数组
result_inplace += 2 * data

import numpy as np

# ❌ 陷阱：切片是视图，修改会影响原数组
original = np.array([1, 2, 3, 4, 5])
subset = original[2:4]       # 这是视图，不是副本！
subset[0] = 999
print(original)              # [  1   2 999   4   5] ← 原数组被改了！

# ✅ 解决：显式复制
original = np.array([1, 2, 3, 4, 5])
subset = original[2:4].copy()  # 创建真正的副本
subset[0] = 999
print(original)              # [1 2 3 4 5] ← 原数组不受影响

import numpy as np

# ❌ 错误：直接比较浮点数
result = 0.1 + 0.2
print(f"0.1 + 0.2 = {result}")        # 0.30000000000000004
print(f"== 0.3? {result == 0.3}")     # False！

# ✅ 正确：使用 isclose
print(f"isclose? {np.isclose(result, 0.3)}")  # True

# 批量比较
arr = np.array([0.1 + 0.2, 0.5 + 0.5, 1.0 / 3.0 * 3])
targets = np.array([0.3, 1.0, 1.0])
matches = np.isclose(arr, targets)
print(f"批量比较: {matches}")  # [ True  True  True]