股票打板策略分析_打板选股技巧

def main(args: Array[String]): Unit = { 

val data=spark
.read
.option("header", true)
.csv(path)
.select("ts_code","trade_date","open","high","low","close","pre_close","change","pct_chg","vol","amount")
// close 收盘价 pre_close 昨收价 change 涨跌额 pct_chg 涨跌幅 vol 成交量 （手） amount 成交额 （千元）
data.createOrReplaceTempView("trade")
val stocks=spark
.read
.option("header", true)
.csv(stocksPath)
stocks.createOrReplaceTempView("stocks")
val dates=spark
.read
.option("header", true)
.csv(datesPath)
dates.createOrReplaceTempView("dates")
val start=LocalDate.of(2018,1,1)
val loop = new Breaks;
val pattern=DateTimeFormatter.ofPattern("yyyyMMdd")
// 循环的日期
loop.breakable{ 

for(i<- 1 to 2000){ 

val lastDate = start.plusDays(i).format(pattern)
val startDate=LocalDate.parse(lastDate,pattern).plusDays(-30).format(pattern)
lastContinueDays(lastDate,startDate)
if(lastDate.equals("20211104")){ 

loop.break
}
}
}
}
// 算出截止lastDate过去10天内，连续涨停的情况
def lastContinueDays(lastDate:String,startDate:String): Unit ={ 

println(s"业务时间: '${startDate}' '${lastDate}'")
// 首先选出涨停的票
sql(
s""" |select | ts_code,trade_date,$lastDate as end_date |from( | select | ts_code,trade_date,open,high,low,close,pre_close,change,pct_chg,vol,amount,row_number() over(partition by ts_code order by trade_date desc) as rn | from | trade | where | -- 时间要换掉 大致的过滤条件 | trade_date>='${startDate}' | and trade_date<='${lastDate}' |)tmp |where | -- 过去10条记录(这里注意一下不一定是过去10天的) | rn<=10 | -- 涨停的数据 | and pct_chg>=9.8 |""".stripMargin
).createOrReplaceTempView("zhangting")
sql(
""" |select | ts_code,trade_date,end_date,zt_cnt |from( | select | ts_code, | trade_date, | end_date, | zt_cnt, | row_number()over(partition by ts_code order by zt_cnt desc) as rn | -- 筛选出 zt_cnt最大的记录 | from( | select | a.ts_code, | a.trade_date, | a.end_date, | count(distinct b.trade_date) as zt_cnt | from | zhangting a | left join | zhangting b | on | a.ts_code=b.ts_code | and a.trade_date<=b.trade_date | and a.end_date>=b.trade_date | left join | dates dates | on | dates.cal_date>=a.trade_date | and dates.cal_date<=a.end_date | -- 是否是交易日 | and dates.is_open=1 | group by | a.ts_code,a.trade_date,a.end_date | having | count(distinct b.trade_date)=count(distinct dates.cal_date) | )t |)t |where | rn=1 |order by | zt_cnt |""".stripMargin
)
.write
.format("parquet")
.mode(SaveMode.Overwrite)
.save(s"/Users/gemii/Desktop/data/day=${lastDate}")
}

  def daBan(): Unit ={ 

val data=spark
.read
.parquet("/Users/gemii/Desktop/data")
data.createOrReplaceTempView("daban")
sql(
""" |select | zt_cnt, | cnt, | concat(cast(round(cnt/lag(cnt)over(order by zt_cnt)*100,2) as string),"%") rate |from( | select | zt_cnt, | count(ts_code) as cnt | from | daban | group by | zt_cnt |) |order by | zt_cnt |""".stripMargin
).show(2000,false)
}