市集摊位网络通常包含几十个甚至上百个独立摊位，收款设备可能是扫码枪、POS机或摊主自己的手机。每个设备产生的销售流水若不能及时汇聚，运营方就无法判断哪些摊位需要补货、哪些交易存在争议。本文把管理端放在R语言环境里，先通过plumber提供HTTP接口接收摊位数据，再用data.table和SQLite完成清洗、入账与状态跟踪，最后用shiny输出实时看板。整套方案可以部署在一台普通Linux服务器或Windows笔记本上，适合没有专门后端团队的市集运营方。

一、用plumber构建摊位数据接收接口

plumber可以把R函数直接转换成HTTP接口，不需要额外学习框架。摊位采集端只负责把销售记录POST到管理节点的/sales路径，管理端解析JSON后先做字段校验，再放入接收队列。这里建议使用JSON作为数据交换格式，因为它比CSV更容易表达嵌套结构和空值，也比直接拼接URL参数更适合交易场景。

接口收到数据后不能只返回200了事。如果字段缺失、金额不是数字或时间格式错误，应当返回4xx状态码，让采集端明确知道需要重传还是修正。下面给出一个最小可运行的plumber入口文件，它校验五个必需字段：stall_id、amount、item_count、ts、trade_no。trade_no由摊位号加时间戳和随机数组成，用来全局唯一标识一笔交易。

# plumber.R
library(plumber)
library(jsonlite)
#* @post /sales
#* @serializer unboxedJSON
function(req, res) {
  body <- req$postBody
  payload <- tryCatch(fromJSON(body), error = function(e) NULL)
  if (is.null(payload)) {
    res$status <- 400
    return(list(code = 400L, message = "invalid json"))
  }
  required <- c("stall_id", "amount", "item_count", "ts", "trade_no")
  if (!all(required %in% names(payload))) {
    res$status <- 422
    return(list(code = 422L, message = "missing fields"))
  }
  list(code = 200L, message = "accepted")
}

实际部署时，建议把/sales接口做成无状态接收，接到的数据先写入一个轻量队列文件或直接批量插入SQLite，不要在这个接口里做复杂统计。这样可以降低摊位端超时概率。若当天摊位数量超过500个，单进程plumber可能成为瓶颈，可以考虑把plumber挂在Nginx后面，并启动多个R worker，通过负载均衡把请求分散到不同进程。

二、交易状态流转与数据清洗

原始流水只代表采集端认为自己卖出了一单，并不等于管理端已经入账。每笔交易建议维护一个状态字段，至少包含received、validated、posted、failed、review等状态。received表示接口收到，validated表示金额和字段格式通过校验，posted表示已经写入正式账本，review表示触发人工复核。状态机一旦清晰，后续对账和审计都会容易很多。

清洗阶段主要处理三类问题：重复上报、异常金额和时间戳乱序。重复上报通常是因为摊位端网络超时后自动重试，但服务端实际已经写入。可以用trade_no做唯一键，配合INSERT OR IGNORE或唯一索引来去重。异常金额包括小于等于0、超过单笔上限或与商品数量明显不匹配。下面示例用data.table快速去重并打标。

library(data.table)
tx <- data.table(
  trade_no = c("T001", "T001", "T002"),
  stall_id = c("A12", "A12", "B03"),
  amount = c(23.5, 23.5, 18.0),
  ts = c("2025-06-01 10:01:02", "2025-06-01 10:01:02", "2025-06-01 10:03:11")
)
clean_tx <- unique(tx, by = "trade_no")
threshold <- 500
clean_tx[, flag := ifelse(amount <= 0 | amount > threshold, "review", "normal")]
print(clean_tx)

data.table在数据量达到几十万行时优势明显，它的按引用修改不需要反复复制整个data.frame。如果市集有多个活动日，建议按日期分表或至少建立日期索引。时间戳乱序问题可以通过导入时统一转换为UTC，并在看板层转换回本地时区，避免夏令时或跨时区活动造成对账差一小时。

三、并发写入与失败重试机制

多个摊位同时上报时，管理端必须保证同一个trade_no不会被重复处理两次，同时批量插入不能部分成功。SQLite在单机场景足够稳定，开启WAL模式后读写可以并发进行。写入前用dbBegin开启事务，全部执行完成后dbCommit提交，任何一步报错就dbRollback回滚。下面给出建表和写入函数。

library(DBI)
library(RSQLite)
con <- dbConnect(SQLite(), "market.db")
dbExecute(con, "PRAGMA journal_mode=WAL")
dbExecute(con, "CREATE TABLE IF NOT EXISTS transactions (
  trade_no TEXT PRIMARY KEY,
  stall_id TEXT,
  amount REAL,
  state TEXT,
  created_at TEXT
)")
insert_tx <- function(con, row) {
  dbBegin(con)
  tryCatch({
    dbExecute(con, "INSERT OR IGNORE INTO transactions VALUES (?, ?, ?, ?, ?)",
              params = unname(row))
    dbCommit(con)
    TRUE
  }, error = function(e) {
    dbRollback(con)
    FALSE
  })
}

失败重试需要区分错误类型。字段缺失、JSON格式错误这类客户端问题重试没有意义，应当直接返回4xx并记录日志；数据库锁定、磁盘写入慢、网络闪断这类服务端或链路问题可以返回5xx，让采集端按指数退避重试。采集端建议至少保存最近7天的本地待发送数据，按顺序逐条发送，只有收到明确成功响应才删除本地副本。这样即使管理端宕机半小时，恢复后数据仍能完整补齐。

对于交易一致性，除了唯一键，还可以对每日数据进行汇总校验。例如每天闭市后，管理端按摊位计算总笔数和总金额，与采集端上报的汇总字段比对。若不一致，通过trade_no明细逐条查找缺口。这个机制比单纯依赖HTTP接口返回更可靠，因为HTTP成功只代表服务端接受，不一定代表最终进入正式账本。

四、用shiny搭建实时对账看板

运营人员需要的不是一行行流水，而是每个摊位的实时销售额、订单数和待处理异常。shiny可以直接连接SQLite表，并用reactivePoll定时查询记录数，当记录数变化时自动刷新汇总。下面示例展示了一个只读看板的核心结构。

library(shiny)
library(DBI)
library(RSQLite)
con <- dbConnect(SQLite(), "market.db")
ui <- fluidPage(
  titlePanel("市集交易看板"),
  mainPanel(
    tableOutput("summary_table")
  )
)
server <- function(input, output, session) {
  data <- reactivePoll(
    intervalMillis = 3000,
    session = session,
    checkFunc = function() {
      dbGetQuery(con, "SELECT COUNT(*) FROM transactions")[1, 1]
    },
    valueFunc = function() {
      dbGetQuery(con, "SELECT stall_id, SUM(amount) AS revenue, COUNT(*) AS orders
                  FROM transactions
                  WHERE state = '已入账'
                  GROUP BY stall_id")
    }
  )
  output$summary_table <- renderTable(data())
}

看板可以按摊位、小时、支付方式分组展示，并设置异常金额高亮。若需要把看板分享给多个管理者，不建议直接开放shiny server端口，最好在前面加一层认证代理或使用shinyapps.io的私有部署。对于临时市集，也可以导出静态HTML报告，由RMarkdown在活动结束后一键生成。

这套基于R的方案胜在组件少、部署快，所有代码都可以在单个脚本或几个文件中维护。对数据量在百万级以内的中小市集来说，SQLite和data.table完全够用。如果后续要接入更多设备或跨区域市集，可以将接收接口替换为消息队列消费者，并把数据库迁移到PostgreSQL，R代码中的清洗逻辑几乎不用改动。