Scala ZIO로 구현해본 종가매매 종목 추출 자동화

** 현재 작성중입니다 **

 

종가 매매?

• 유튜브 등을 참고하여 종가 매매의 패턴은 아래와 같습니다. (장 마감전에 아래와 같은 종목을 빨리 발견해야 합니다)
  • 오늘 기준 09:00~15:10 시간대의 주가가 우상향 할 것
  • 최근 5일 평균 거래액이 10억 이상, 거래량이 5배 이상 증가
  • 최근 3~6개월을 기준으로 전고점을 강하게 돌파할 것

 

기술적으로는?

• pFP와 ZIO를 공부한다는 느낌으로 진행합니다. 물론 구현의 난이도는 python이 훨씬 낮습니다.
  • Fiber를 활용하여 대량의 데이터 분석을 병렬로 처리할 수 있는 등 ZIO가 가지는 장점을 활용해 보려고 합니다.
• 현재 운용중인 https://www.heyes.live 에서 15:00부터 대상 종목 제공하는 것을 목표로 합니다.
  • Web기반의 API를 제공하는 모듈로 배포됩니다.

 

 

크롤링

 

• 네이버 Pay증권의 화면을 crawling 하여 필요한 정보를 추출후 db에 저장하는 형태가 됩니다.
• 주어진 url인 https://finance.naver.com/item/sise_time.naver?code=000660&thistime=20250115124308&page=1 는 10분 단위로 분당 거래량, 체결가 등의 정보가 제공됩니다.

• 주의할 점은 네이버와 같이 공개된 페이지를 크롤링할 경우 과도하게 요청 트래픽이 발생하지 않도록 1tps 이상의 저속크롤링을 유지해야 된다는 점입니다.
• 일단은 해당 페이지는 10분 주기로 데이터 수집이 이루어져야 하기 때문에 별도의 스케쥴링 전략이 필요합니다.
  • 우선순위 큐를 만들어 큐에 등록된 seed는 1초 간격으로 하나씩 처리하고 과도하게 pending 상태의 seed job이 발생할 경우 신규 큐를 생성하여 별도의 worker thread가 처리 속도를 높이는 식으로 일단은 구현을 합니다.
  • 아래는 ZIOf로 해당 큐기반의 worker scheduler를 구현한 코드입니다.

abstract class UnitJobScheduler (
  queueRef: Ref[List[Queue[String]]], 
  queueSize: Int = 10) extends JobProducer[String] {

  private def inputDataAndExtendQueue(queue: List[Queue[String]])(effect : => String): ZIO[Any, Nothing, Boolean] = 
    for {
      result <- ZIO.suspendSucceed {
      queue match {
        case Nil => ZIO.succeed(false)
        case x :: xs => 
          for {
            result <- x.offer(effect)
            res <- result match {
              case true => ZIO.succeed(true)
              case false => 
                inputDataAndExtendQueue(xs)(effect)
            }
          } yield res
        }
      }
    } yield result
  
  def processJob(queue: Queue[String], workerId: Int = -1): ZIO[Any, Nothing, Unit]

  // add new job  
  def addJob(job: String) = 
    for {
      queue <- queueRef.get
      addRes <- inputDataAndExtendQueue(queue)(job)
      _ <- addRes match {
        case true => ZIO.unit
        case false => for {
          _ <- Console.printLine("--> create new worker queue ... " + queue.size)
          q <- Queue.dropping[String](queueSize)
          _ <- queueRef.update(_ :+ q)
          _ <- processJob(q, queue.size).fork
        } yield ()
      }
    } yield ()
  
  // produce continouse jobs
  def startProduce() = {
    (for {
      //TODO check this time is whether holiday or not
      data <- unitProduce // ----------------> Producing Job
      _ <- Console.printLine("produce job to crawl: " + data)
      _ <- ZIO.foreach(data){r => addJob(r)}
    } yield ()).repeat(Schedule.spaced(1.seconds)).unit  
  } 
}

 

• 아래는 위의 코드를 기반으로 Crawler를 구현한 코드입니다.

trait JobProducer[T] {
  def unitProduce: ZIO[Any, Nothing, List[T]]
}

class CrawlJobScheduler(
  queueRef: Ref[List[Queue[String]]], 
  stockRepo: StockRepo, 
  crawler: MinStockCrawler,
  crawlStatus: CrawlStatusRepo) extends UnitJobScheduler(queueRef) {

  override def unitProduce: ZIO[Any, Nothing, List[String]] = (for {
    _ <- ZIO.when(DataUtil.getCurrentTimestamp() != DataUtil.stockTimestamp(0))(
      ZIO.log("Out of stockTime") *> ZIO.fail(new Exception("---> Out of stock time of Today")))
    queue <- queueRef.get
    cntEndItemCodes <- queue.lastOption match {
      case Some(q) => q.size
      case None => ZIO.succeed(0)
    } 
    expiredItemCodes <- crawlStatus.getExpiredItemCode(10 * 60 * 1000 - cntEndItemCodes * 1000)
    _ <- ZIO.foreachPar(expiredItemCodes){itemCode => crawlStatus.syncCrawlStatus(itemCode, "PEND")}
    _ <- ZIO.log("Create target itemCodes : " + expiredItemCodes)
  } yield expiredItemCodes)
  .catchAll(e => ZIO.log(e.getMessage()) *> ZIO.succeed(List.empty[String]))

  override def processJob(queue: Queue[String], workerId: Int = -1): ZIO[Any, Nothing, Unit] = 
    (for {
      itemCode <- queue.take
      cd <- crawler.crawl(itemCode)
      res <- stockRepo.insertStockMinVolumeSerialBulk(cd)
      _ <- crawlStatus.syncCrawlStatus(itemCode, "SUSP")
      _ <- ZIO.foreach(cd){r => Console.printLine(r.toString())}
      _ <- Console.printLine(s"Inserted data ${cd.size} for $itemCode by worker #$workerId").ignore // log
    } yield ()).repeat(Schedule.spaced(1.seconds))
    .provide(
      Client.customized,
      NettyClientDriver.live,
      ZLayer.succeed(NettyConfig.default),
      DnsResolver.default,
      DataDownloader.live,
    )
    .unit.catchAll(e => ZIO.log(e.getMessage()).ignore)
}

 

 

종가패턴 발견

• 우상향 여부를 판별하고 기울기 값과 오차범위내의 가격을 count하여 패턴의 일치여부를 찾습니다.
•  

class EndPriceAnalyzerImpl extends EndPriceAnalyzer {

  def analyzeUpstream(data: List[StockMinVolumeTable],  allowedError: Double = 0.25, innerErrorPer: Double = 0.80) = {
        
    val streamedData = ZStream.fromIterable(data).filter(r => {
        val tokens = r.tsCode.split("_")
        tokens.length == 2 && tokens(1) <= "15:30"
      })
    
    (for {
      firstElem <- streamedData.runHead
      lastElem <- streamedData.runLast
      _ <- Console.printLine(s"First : ${firstElem}, end : ${lastElem}")
      delta <- ZIO.attempt(
        lastElem.flatMap(l => firstElem.map(f => l.fixedPrice - f.fixedPrice))
        .getOrElse(0.0)
        )
      _ <- Console.printLine(s"Delta : ${delta}")
      _ <- ZIO.when(delta < 0)(ZIO.fail(new Exception("---> Stock price is not trending upward <--")))
      slope <- ZIO.attempt(delta / (data.length - 1))
      _ <- Console.printLine(s"Slope : ${slope}")

      result <- streamedData.zipWithIndex.runFold(0) { case (inCount, (element, idx)) =>
        val expectedY = firstElem.map(a => a.fixedPrice + slope * idx) 
        val error = expectedY.map(exp => Math.abs(element.fixedPrice - exp) / delta)
        error match {
          case Some(e) if e <= allowedError => 
            inCount + 1
          case x => 
            inCount
        }
      }
      _ <- Console.printLine(s"Result : ${result} / ${data.length}")
      finResult <- ZIO.succeed(result > data.length * innerErrorPer)
    } yield finResult).catchAll(e => Console.printError(e.getLocalizedMessage()) *> ZIO.succeed(false))
    
  }

  override def analyze(stockCode: String, targetDt: String) = for {
    repo <- ZIO.service[StockRepo]
    targetData <- repo.selectStockDataByItemCode(stockCode, targetDt)
    result <- analyzeUpstream(targetData)
  } yield result
  
}